223 lines
12 KiB
PHP
223 lines
12 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\common\service\prescriptionai;
|
|
|
|
use app\common\model\doctor\Medicine;
|
|
use think\facade\Db;
|
|
|
|
final class PrescriptionAiWorker
|
|
{
|
|
/** The coordinator does no model work; run one worker for each model separately. */
|
|
public function prepareOne(): bool
|
|
{
|
|
if (!PrescriptionAiStore::enabled() || !($batch = PrescriptionAiStore::claimBatch())) {
|
|
return false;
|
|
}
|
|
try {
|
|
$actor = PrescriptionAiAccess::actor((int) $batch['actor_id']);
|
|
$rx = $actor ? PrescriptionAiAccess::prescription((int) $batch['prescription_id'], (int) $batch['actor_id'], $actor) : null;
|
|
if (!$actor || !$rx || !PrescriptionAiAccess::canGenerate((int) $batch['actor_id'], $actor)) {
|
|
PrescriptionAiStore::releasePreparation($batch, ['status' => 'blocked', 'error_code' => 'ACCESS_REVOKED']);
|
|
return true;
|
|
}
|
|
if (!PrescriptionAiPolicy::isManual($rx) || !hash_equals($batch['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
|
|
PrescriptionAiStore::invalidate((int) $batch['prescription_id'], 'prescription_changed');
|
|
return true;
|
|
}
|
|
$frozenRx = (new PrescriptionAiCipher())->decrypt($batch['prescription_cipher'], 'prescription');
|
|
$context = Db::transaction(static fn (): array => PrescriptionAiContext::build(
|
|
$frozenRx, (int) $batch['actor_id'], $actor, (int) $batch['decision_at']));
|
|
$context['actor_scope_hash'] = self::scopeHash((int) $batch['actor_id'], $actor);
|
|
// Freeze one complete identity dictionary for both branches, outside model evidence.
|
|
$context['_comparison_catalog'] = Medicine::where('status', 1)->whereNull('delete_time')
|
|
->field(['id', 'name', 'unit'])->order('id')->select()->toArray();
|
|
$context['dictionary_version'] = hash('sha256', PrescriptionAiPolicy::canonical($context['_comparison_catalog']));
|
|
if (strlen(PrescriptionAiPolicy::canonical($context)) > (int) config('prescription_analysis.max_context_bytes', 8000000)) {
|
|
PrescriptionAiStore::releasePreparation($batch, ['status' => 'blocked', 'error_code' => 'CONTEXT_TOO_LARGE']);
|
|
return true;
|
|
}
|
|
if (!empty($context['wait_for_transcript']) && time() >= (int) $batch['wait_until']) {
|
|
$context['missing'][] = ['code' => 'TRANSCRIPT_NOT_FINAL', 'message' => '本次问诊转写尚未完整归档'];
|
|
$context['baseline_eligible'] = false;
|
|
$context['baseline_exclusion_reasons'][] = 'transcript_not_final';
|
|
}
|
|
PrescriptionAiStore::finishPreparation($batch, $context);
|
|
} catch (\Throwable $e) {
|
|
$attempt = (int) $batch['prepare_attempts'] + 1;
|
|
PrescriptionAiStore::releasePreparation($batch, [
|
|
'status' => $attempt >= 3 ? 'blocked' : 'retry_wait', 'prepare_attempts' => $attempt,
|
|
'error_code' => 'SOURCE_PREPARATION_FAILED', 'next_run_at' => time() + 60,
|
|
]);
|
|
}
|
|
return true;
|
|
}
|
|
|
|
public function runOne(string $model): bool
|
|
{
|
|
if (!PrescriptionAiStore::enabled() || !($task = PrescriptionAiStore::claimTask($model))) {
|
|
return false;
|
|
}
|
|
try {
|
|
$batch = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->find();
|
|
$cipher = new PrescriptionAiCipher();
|
|
$context = $cipher->decrypt((string) $batch['context_cipher'], 'context');
|
|
if (!$this->authorizedAndCurrent($batch, $context)) {
|
|
PrescriptionAiStore::fail($task, 'ACCESS_REVOKED', false);
|
|
return true;
|
|
}
|
|
if (!empty($task['progress_cipher'])) {
|
|
$context['_progress'] = $cipher->decrypt($task['progress_cipher'], 'progress:' . $task['id']);
|
|
}
|
|
$public = [];
|
|
$checkpoint = function (array $progress, bool $persistCache = true) use ($task, $batch, $context, &$public): bool {
|
|
if (!PrescriptionAiStore::enabled() || !$this->authorizedAndCurrent($batch, $context)
|
|
|| !PrescriptionAiStore::checkpoint($task, $progress, $persistCache)) {
|
|
return false;
|
|
}
|
|
$public = PrescriptionAiProgress::sanitize($progress['public'] ?? null);
|
|
return true;
|
|
};
|
|
$output = PrescriptionAiGenerator::generate($model, $context, $checkpoint);
|
|
if (empty($output['ok'])) {
|
|
PrescriptionAiStore::fail($task, (string) ($output['error_code'] ?? 'UPSTREAM_FAILED'), !empty($output['retryable']));
|
|
return true;
|
|
}
|
|
if (!$this->authorizedAndCurrent($batch, $context)) {
|
|
PrescriptionAiStore::fail($task, 'SOURCE_CHANGED', false);
|
|
return true;
|
|
}
|
|
$doctor = $cipher->decrypt($batch['prescription_cipher'], 'prescription');
|
|
$doctor['herbs'] = PrescriptionAiPolicy::decode($doctor['herbs'] ?? []);
|
|
$doctor['aux_usage'] = PrescriptionAiPolicy::decode($doctor['aux_usage'] ?? []);
|
|
$candidate = is_array($output['candidate'] ?? null) ? $output['candidate'] : [];
|
|
$catalog = (array) ($context['_comparison_catalog'] ?? []);
|
|
if (!$checkpoint(['public' => PrescriptionAiProgress::advance($public, 'comparing')], false)) {
|
|
PrescriptionAiStore::fail($task, 'CHECKPOINT_REJECTED', false);
|
|
return true;
|
|
}
|
|
$comparison = PrescriptionAiComparison::compare($doctor, $candidate, $catalog);
|
|
$comparison['dictionary_version'] = (string) ($context['dictionary_version'] ?? '');
|
|
PrescriptionAiStore::complete($task, $output, $comparison);
|
|
} catch (\Throwable $e) {
|
|
// No PHI, URLs, SQL or upstream body in logs / error messages.
|
|
PrescriptionAiStore::fail($task, 'INTERNAL_ERROR', true);
|
|
}
|
|
return true;
|
|
}
|
|
|
|
private function authorizedAndCurrent(array $batch, array $context): bool
|
|
{
|
|
$fresh = Db::name('prescription_ai_batch')->where('id', $batch['id'])->find();
|
|
$actor = PrescriptionAiAccess::actor((int) $batch['actor_id']);
|
|
if (!$fresh || $fresh['validity'] !== 'current' || !$actor
|
|
|| !PrescriptionAiAccess::canGenerate((int) $batch['actor_id'], $actor)
|
|
|| !hash_equals((string) ($context['actor_scope_hash'] ?? ''), self::scopeHash((int) $batch['actor_id'], $actor))
|
|
|| !PrescriptionAiAccess::sourceIds($context['source_diagnosis_ids'] ?? [], (int) $batch['actor_id'], $actor)) {
|
|
return false;
|
|
}
|
|
if (!PrescriptionAiContext::assertSnapshotAccess($context, (int) $batch['actor_id'], $actor)) {
|
|
return false;
|
|
}
|
|
$rx = PrescriptionAiAccess::prescription((int) $batch['prescription_id'], (int) $batch['actor_id'], $actor);
|
|
return $rx && PrescriptionAiPolicy::isManual($rx)
|
|
&& hash_equals((string) $batch['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx));
|
|
}
|
|
|
|
public static function scopeHash(int $adminId, array $actor): string
|
|
{
|
|
$permissions = \app\adminapi\logic\auth\AuthLogic::getAuthByAdminId($adminId);
|
|
sort($permissions);
|
|
$roles = (array) ($actor['role_id'] ?? []);
|
|
$departments = (array) ($actor['dept_id'] ?? []);
|
|
sort($roles);
|
|
sort($departments);
|
|
return hash('sha256', PrescriptionAiPolicy::canonical([
|
|
'admin_id' => $adminId, 'root' => $actor['root'] ?? 0,
|
|
'roles' => $roles, 'departments' => $departments, 'permissions' => $permissions,
|
|
'scope' => \app\common\service\DataScope\DataScopeService::getEffectiveScope($actor),
|
|
]));
|
|
}
|
|
|
|
/** Bounded reconciliation of missing save events, also catches old clients. */
|
|
public function reconcile(int $afterId = 0, int $limit = 50, ?int $from = null, ?int $to = null, bool $apply = true): array
|
|
{
|
|
$from = $from ?? (int) config('prescription_analysis.start_at', 0);
|
|
if ($from <= 0) {
|
|
return ['selected' => 0, 'enqueued' => 0, 'last_id' => $afterId, 'reason' => 'start_at_required'];
|
|
}
|
|
$query = Db::name('tcm_prescription')->where('id', '>', $afterId)->where('is_system_auto', 0)
|
|
->where('void_status', 0)->whereNull('delete_time')->where('update_time', '>=', $from);
|
|
if ($to !== null) {
|
|
$query->where('update_time', '<=', $to);
|
|
}
|
|
$rows = $query->order('id')->limit(max(1, min(200, $limit)))->select()->toArray();
|
|
$enqueued = 0;
|
|
foreach ($rows as $rx) {
|
|
$afterId = (int) $rx['id'];
|
|
if (!PrescriptionAiPolicy::isManual($rx)) {
|
|
continue;
|
|
}
|
|
$actor = PrescriptionAiAccess::actor((int) $rx['creator_id']);
|
|
if (!$actor || !PrescriptionAiAccess::canGenerate((int) $rx['creator_id'], $actor)
|
|
|| !PrescriptionAiAccess::prescription($afterId, (int) $rx['creator_id'], $actor)) {
|
|
continue;
|
|
}
|
|
$subject = Db::name('prescription_ai_subject')->where('prescription_id', $afterId)->find();
|
|
if ($subject && hash_equals($subject['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
|
|
continue;
|
|
}
|
|
if ($apply) {
|
|
Db::transaction(static function () use ($rx, $actor): void {
|
|
$fresh = Db::name('tcm_prescription')->where('id', $rx['id'])->lock(true)->find();
|
|
if ($fresh) {
|
|
PrescriptionAiStore::recordSaved($fresh, (int) $rx['creator_id'], $actor, ['trigger' => 'reconciled']);
|
|
}
|
|
});
|
|
}
|
|
$enqueued++;
|
|
}
|
|
return ['selected' => count($rows), 'enqueued' => $enqueued, 'last_id' => $afterId];
|
|
}
|
|
|
|
/** Detects late notes, reports, daily records and transcripts without client-side generation. */
|
|
public function refreshSources(int $afterId = 0, int $limit = 10): array
|
|
{
|
|
if (!config('prescription_analysis.auto_refresh_sources', true)) {
|
|
return ['selected' => 0, 'last_id' => $afterId, 'enqueued' => 0];
|
|
}
|
|
$cutoff = time() - max(1, (int) config('prescription_analysis.refresh_recent_days', 7)) * 86400;
|
|
$rows = Db::name('prescription_ai_batch')->where('id', '>', $afterId)->where('validity', 'current')
|
|
->whereIn('status', ['success', 'partial', 'failed'])->where('created_at', '>=', $cutoff)
|
|
->order('id')->limit(max(1, min(30, $limit)))->select()->toArray();
|
|
$enqueued = 0;
|
|
foreach ($rows as $batch) {
|
|
$afterId = (int) $batch['id'];
|
|
try {
|
|
$actor = PrescriptionAiAccess::actor((int) $batch['actor_id']);
|
|
$rx = $actor ? PrescriptionAiAccess::prescription((int) $batch['prescription_id'], (int) $batch['actor_id'], $actor) : null;
|
|
if (!$actor || !$rx || !PrescriptionAiAccess::canGenerate((int) $batch['actor_id'], $actor) || !PrescriptionAiPolicy::isManual($rx)) {
|
|
continue;
|
|
}
|
|
$context = PrescriptionAiContext::build($rx, (int) $batch['actor_id'], $actor, (int) $batch['decision_at']);
|
|
if (($context['source_hash'] ?? '') === $batch['source_hash']) {
|
|
continue;
|
|
}
|
|
$todayCount = Db::name('prescription_ai_batch')->where('patient_id', $batch['patient_id'])
|
|
->where('created_at', '>=', strtotime('today'))->count();
|
|
if ($todayCount >= (int) config('prescription_analysis.daily_patient_batches', 10)) {
|
|
continue;
|
|
}
|
|
PrescriptionAiStore::enqueue($rx, (int) $batch['actor_id'], 'source_update',
|
|
hash('sha256', 'source:' . $rx['id'] . ':' . $batch['prescription_revision'] . ':' . $context['source_hash']),
|
|
['reason' => '患者资料或问诊转写已更新']);
|
|
$enqueued++;
|
|
} catch (\Throwable $e) {
|
|
// Existing report remains available; a later bounded sweep retries the source read.
|
|
}
|
|
}
|
|
return ['selected' => count($rows), 'last_id' => $afterId, 'enqueued' => $enqueued];
|
|
}
|
|
}
|