368 lines
21 KiB
PHP
368 lines
21 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\adminapi\logic\tcm;
|
|
|
|
use app\common\service\mobilerecording\MobileRecordingStorage as Storage;
|
|
use app\common\service\mobilerecording\MobileRecordingStream;
|
|
use app\common\service\mobilerecording\MobileRecordingResyncException;
|
|
use app\common\service\mobilerecording\MobileRecordingLocalUnavailableException;
|
|
use app\common\service\mobilerecording\MobileRecordingObjects as Objects;
|
|
use DomainException;
|
|
use think\facade\Db;
|
|
|
|
/** The binding is the authorization anchor; upload IDs are not bearer credentials. */
|
|
final class MobileRecordingMediaLogic
|
|
{
|
|
public const CHUNK_SIZE = 1048576;
|
|
public const AUDIO_MAX = 536870912;
|
|
public const TEXT_MAX = 8388608;
|
|
private const MIMES = ['audio/wav', 'audio/flac', 'audio/ogg', 'audio/mpeg', 'audio/mp4', 'audio/amr'];
|
|
|
|
public static function prepare(array $p, int $actor): array
|
|
{
|
|
MobileRecordingLogic::only($p, ['installation_id', 'recording_key', 'diagnosis_id', 'kind', 'size_bytes', 'sha256', 'media_type']);
|
|
$identity = MobileRecordingLogic::identity($p);
|
|
$diagnosis = self::diagnosisId($p);
|
|
$kind = $p['kind'] ?? '';
|
|
if (!in_array($kind, ['audio', 'transcript'], true)) { throw new DomainException('上传类型无效'); }
|
|
$size = MobileRecordingLogic::integer($p['size_bytes'] ?? null, $kind === 'audio' ? 1 : 0,
|
|
$kind === 'audio' ? self::AUDIO_MAX : self::TEXT_MAX, '文件大小');
|
|
$sha = self::hash($p['sha256'] ?? null);
|
|
$mime = $p['media_type'] ?? '';
|
|
if (!is_string($mime) || ($kind === 'audio' ? !in_array($mime, self::MIMES, true) : $mime !== 'text/plain')) {
|
|
throw new DomainException('文件媒体类型无效');
|
|
}
|
|
return self::mutate(static function (array &$garbage) use ($identity, $diagnosis, $kind, $size, $sha, $mime, $actor): array {
|
|
$binding = self::owned($identity, $diagnosis, $actor);
|
|
$id = (int) $binding['id'];
|
|
$existing = Db::name('mobile_recording_upload')->where('binding_id', $id)->where('kind', $kind)->where('sha256', $sha)->find();
|
|
if ($existing && ((int) $existing['size_bytes'] !== $size || $existing['media_type'] !== $mime)) {
|
|
throw new DomainException('相同摘要的文件信息不一致');
|
|
}
|
|
// Only one unfinished version per kind; the published version stays readable until replacement commits.
|
|
$obsolete = Db::name('mobile_recording_upload')->where('binding_id', $id)->where('kind', $kind)
|
|
->where('state', 'uploading')->where('sha256', '<>', $sha)->column('upload_id');
|
|
self::deleteUploads($obsolete, $garbage);
|
|
if ($existing) {
|
|
if ($existing['state'] === 'complete') {
|
|
try { Objects::verifyAvailable($existing); }
|
|
catch (MobileRecordingLocalUnavailableException $e) {
|
|
Db::name('mobile_recording_asset')->where('upload_id', $existing['upload_id'])->delete();
|
|
$existing['state'] = 'uploading'; $existing['received_bytes'] = 0;
|
|
Db::name('mobile_recording_upload')->where('upload_id', $existing['upload_id'])
|
|
->update(['state' => 'uploading', 'received_bytes' => 0, 'updated_at' => time()]);
|
|
}
|
|
}
|
|
return self::dto(self::synchronize($existing));
|
|
}
|
|
$row = ['upload_id' => bin2hex(random_bytes(16)), 'binding_id' => $id, 'diagnosis_id' => $diagnosis,
|
|
'kind' => $kind, 'sha256' => $sha, 'media_type' => $mime, 'size_bytes' => $size,
|
|
'received_bytes' => 0, 'state' => 'uploading', 'created_at' => time(), 'updated_at' => time()];
|
|
Db::name('mobile_recording_upload')->insert($row);
|
|
Storage::locked($row['upload_id'], static fn ($file): int => Storage::reconcile($file, 0));
|
|
return self::dto($row);
|
|
});
|
|
}
|
|
|
|
public static function chunk(array $p, string $bytes, int $actor): array
|
|
{
|
|
MobileRecordingLogic::only($p, ['upload_id', 'diagnosis_id', 'offset', 'chunk_sha256']);
|
|
$uploadId = self::uploadId($p['upload_id'] ?? null);
|
|
$diagnosis = self::diagnosisId($p);
|
|
$offset = MobileRecordingLogic::integer($p['offset'] ?? null, 0, self::AUDIO_MAX, '分片偏移');
|
|
$sha = self::hash($p['chunk_sha256'] ?? null);
|
|
$length = strlen($bytes);
|
|
if ($length < 1 || $length > self::CHUNK_SIZE) { throw new DomainException('分片大小必须为 1 至 1048576 字节'); }
|
|
if (!hash_equals($sha, hash('sha256', $bytes))) { throw new DomainException('分片摘要校验失败'); }
|
|
return self::mutate(static function (array &$garbage) use ($uploadId, $diagnosis, $offset, $bytes, $length, $actor): array {
|
|
$row = self::session($uploadId, $diagnosis, $actor);
|
|
if ($row['state'] === 'complete') {
|
|
if ($offset + $length > (int) $row['size_bytes']) { throw new DomainException('分片超出声明的文件大小'); }
|
|
$source = Objects::source($row);
|
|
try { $existing = is_callable($source) ? $source($offset, $length) : Storage::bytes($source, $offset, $length); }
|
|
finally { if (is_resource($source)) { fclose($source); } }
|
|
if (!hash_equals($bytes, $existing)) { throw new MobileRecordingResyncException('分片与已接收内容冲突，请查询上传状态'); }
|
|
return self::dto($row);
|
|
}
|
|
return Storage::locked($uploadId, static function ($file) use ($row, $offset, $bytes, $length): array {
|
|
$received = Storage::reconcile($file, (int) $row['received_bytes']);
|
|
if ($offset + $length > (int) $row['size_bytes']) { throw new DomainException('分片超出声明的文件大小'); }
|
|
if ($offset < $received && $offset + $length <= $received) {
|
|
if (!hash_equals($bytes, Storage::bytes($file, $offset, $length))) {
|
|
throw new MobileRecordingResyncException('分片与已接收内容冲突，请查询上传状态');
|
|
}
|
|
} elseif ($offset === $received && $row['state'] === 'uploading') {
|
|
Storage::write($file, $offset, $bytes);
|
|
$received += $length;
|
|
} else { throw new MobileRecordingResyncException('分片偏移不连续，请查询上传状态'); }
|
|
Db::name('mobile_recording_upload')->where('upload_id', $row['upload_id'])
|
|
->update(['received_bytes' => $received, 'updated_at' => time()]);
|
|
return self::dto(array_replace($row, ['received_bytes' => $received]));
|
|
});
|
|
});
|
|
}
|
|
|
|
public static function complete(array $p, int $actor): array
|
|
{
|
|
MobileRecordingLogic::only($p, ['upload_id', 'diagnosis_id']);
|
|
$uploadId = self::uploadId($p['upload_id'] ?? null);
|
|
$diagnosis = self::diagnosisId($p);
|
|
// Persist the object location and encryption key before the first COS call.
|
|
$prepared = self::mutate(static function (array &$garbage) use ($uploadId, $diagnosis, $actor): array {
|
|
$row = self::session($uploadId, $diagnosis, $actor);
|
|
if ($row['state'] === 'complete') { return self::dto($row); }
|
|
if (Objects::ready($row)) { return ['prepared' => true]; }
|
|
$row = self::synchronize($row);
|
|
if ((int) $row['received_bytes'] !== (int) $row['size_bytes']) { throw new MobileRecordingResyncException('文件尚未上传完整'); }
|
|
$error = Storage::locked($uploadId, static function ($file) use ($row): ?string {
|
|
try { Storage::validate($file, $row); return null; }
|
|
catch (\DomainException $e) {
|
|
if (!ftruncate($file, 0) || !fflush($file)) { throw new \RuntimeException('Upload reset failed'); }
|
|
return $e->getMessage();
|
|
}
|
|
});
|
|
if ($error !== null) {
|
|
Db::name('mobile_recording_upload')->where('upload_id', $uploadId)->update(['received_bytes' => 0, 'updated_at' => time()]);
|
|
return ['validation_error' => $error];
|
|
}
|
|
Objects::ensure($row);
|
|
return ['prepared' => true];
|
|
});
|
|
if (!isset($prepared['prepared'])) {
|
|
if (($prepared['state'] ?? '') === 'complete') { self::tryCleanupStaging($uploadId); }
|
|
return $prepared;
|
|
}
|
|
Objects::advance($uploadId, static fn (): array => self::session($uploadId, $diagnosis, $actor));
|
|
$result = self::mutate(static function (array &$garbage) use ($uploadId, $diagnosis, $actor): array {
|
|
$row = self::session($uploadId, $diagnosis, $actor);
|
|
if (!Objects::ready($row)) { throw new \RuntimeException('Private cloud confirmation pending'); }
|
|
if ($row['state'] !== 'complete') { self::publish($row, $garbage); }
|
|
return self::dto(array_replace($row, ['state' => 'complete', 'received_bytes' => (int) $row['size_bytes']]));
|
|
});
|
|
self::tryCleanupStaging($uploadId);
|
|
return $result;
|
|
}
|
|
|
|
private static function tryCleanupStaging(string $id): void
|
|
{
|
|
// Publication is already durable; failure must not unpublish it. The worker retries this task.
|
|
try { Objects::cleanupStaging($id); } catch (\Throwable $ignored) {}
|
|
}
|
|
|
|
private static function publish(array $row, array &$garbage): void
|
|
{
|
|
$uploadId = $row['upload_id'];
|
|
Db::name('mobile_recording_upload')->where('upload_id', $uploadId)
|
|
->update(['state' => 'complete', 'received_bytes' => $row['size_bytes'], 'updated_at' => time()]);
|
|
$asset = ['binding_id' => $row['binding_id'], 'kind' => $row['kind'], 'upload_id' => $uploadId, 'updated_at' => time()];
|
|
$query = Db::name('mobile_recording_asset')->where('binding_id', $row['binding_id'])->where('kind', $row['kind']);
|
|
if ((clone $query)->find()) { $query->update($asset); } else { Db::name('mobile_recording_asset')->insert($asset); }
|
|
$oldIds = Db::name('mobile_recording_upload')->where('binding_id', $row['binding_id'])->where('kind', $row['kind'])
|
|
->where('upload_id', '<>', $uploadId)->column('upload_id');
|
|
self::deleteUploads($oldIds, $garbage);
|
|
}
|
|
|
|
/** Privileged maintenance only, never routed by the HTTP controller. */
|
|
public static function migrateLocal(string $id): void
|
|
{
|
|
$guard = static function () use ($id): array {
|
|
$peek = Db::name('mobile_recording_upload')->where('upload_id', $id)->find();
|
|
$binding = $peek ? Db::name('mobile_recording_binding')->where('id', $peek['binding_id'])->lock(true)->find() : null;
|
|
$row = Db::name('mobile_recording_upload')->where('upload_id', $id)->where('state', 'complete')->lock(true)->find();
|
|
if (!$binding || !$row) { throw new \DomainException('Legacy media reference no longer active'); }
|
|
return $row;
|
|
};
|
|
Db::transaction(static function () use ($id, $guard): void {
|
|
$row = $guard();
|
|
if (Objects::ready($row)) { return; }
|
|
Storage::locked($id, static fn ($file) => Storage::validate($file, $row));
|
|
Objects::ensure($row);
|
|
});
|
|
Objects::advance($id, $guard);
|
|
Objects::cleanupStaging($id);
|
|
}
|
|
|
|
public static function status(array $p, int $actor): array
|
|
{
|
|
MobileRecordingLogic::only($p, ['installation_id', 'recording_key', 'diagnosis_id']);
|
|
$identity = MobileRecordingLogic::identity($p);
|
|
$diagnosis = self::diagnosisId($p);
|
|
return self::mutate(static function (array &$garbage) use ($identity, $diagnosis, $actor): array {
|
|
$binding = self::owned($identity, $diagnosis, $actor);
|
|
$result = [];
|
|
foreach (['audio', 'transcript'] as $kind) {
|
|
// A pending replacement is the mobile recovery target, even while the old asset remains published.
|
|
$rows = Db::name('mobile_recording_upload')->where('binding_id', $binding['id'])->where('kind', $kind)
|
|
->order('state', 'desc')->select()->toArray();
|
|
$row = $rows ? self::synchronize($rows[0]) : null;
|
|
$result[$kind] = $row ? self::assetState($row) : ['state' => 'missing', 'upload_id' => '', 'size_bytes' => 0,
|
|
'received_bytes' => 0, 'sha256' => '', 'media_type' => ''];
|
|
}
|
|
return $result;
|
|
});
|
|
}
|
|
|
|
public static function audio(array $p, int $actor, string $range = ''): MobileRecordingStream
|
|
{
|
|
return self::read($p, $actor, 'audio', static function (array $row) use ($range): MobileRecordingStream {
|
|
return new MobileRecordingStream(Objects::source($row), (int) $row['size_bytes'], $row['media_type'], $range);
|
|
});
|
|
}
|
|
|
|
public static function transcript(array $p, int $actor): array
|
|
{
|
|
return self::read($p, $actor, 'transcript', static function (array $row): array {
|
|
$source = Objects::source($row);
|
|
try { $text = is_callable($source) ? $source(0, (int) $row['size_bytes']) : Storage::bytes($source, 0, (int) $row['size_bytes']); }
|
|
finally { if (is_resource($source)) { fclose($source); } }
|
|
if (!hash_equals($row['sha256'], hash('sha256', $text))) { throw new \RuntimeException('Transcript integrity mismatch'); }
|
|
return ['text' => $text, 'sha256' => $row['sha256'], 'size_bytes' => (int) $row['size_bytes']];
|
|
});
|
|
}
|
|
|
|
private static function read(array $p, int $actor, string $kind, callable $read)
|
|
{
|
|
MobileRecordingLogic::only($p, ['binding_id', 'diagnosis_id']);
|
|
$id = MobileRecordingLogic::integer($p['binding_id'] ?? null, 1, PHP_INT_MAX, '关联标识');
|
|
$anchor = isset($p['diagnosis_id']) ? self::diagnosisId($p) : null;
|
|
return Db::transaction(static function () use ($id, $anchor, $actor, $kind, $read) {
|
|
$info = MobileRecordingLogic::actor($actor, true);
|
|
$binding = Db::name('mobile_recording_binding')->where('id', $id)->lock(true)->find();
|
|
if (!$binding || ($anchor !== null && (int) $binding['diagnosis_id'] !== $anchor)) {
|
|
throw new DomainException('录音关联已变更或不可访问');
|
|
}
|
|
MobileRecordingLogic::diagnosis((int) $binding['diagnosis_id'], $actor, $info, true);
|
|
$asset = Db::name('mobile_recording_asset')->where('binding_id', $id)->where('kind', $kind)->find();
|
|
$row = $asset ? Db::name('mobile_recording_upload')->where('upload_id', $asset['upload_id'])
|
|
->where('binding_id', $id)->where('kind', $kind)->where('state', 'complete')->find() : null;
|
|
if (!$row) { throw new DomainException('此录音尚未上传对应内容'); }
|
|
return $read($row);
|
|
});
|
|
}
|
|
|
|
/** Called only after the binding's actor/diagnoses are locked and authorized. */
|
|
public static function bindingMoved(int $id, int $diagnosisId): array
|
|
{
|
|
$garbage = Db::name('mobile_recording_upload')->where('binding_id', $id)->where('state', 'uploading')->column('upload_id');
|
|
Objects::queueDelete($garbage);
|
|
if ($garbage) { Db::name('mobile_recording_upload')->whereIn('upload_id', $garbage)->delete(); }
|
|
Db::name('mobile_recording_upload')->where('binding_id', $id)->where('state', 'complete')
|
|
->update(['diagnosis_id' => $diagnosisId, 'updated_at' => time()]);
|
|
return $garbage;
|
|
}
|
|
|
|
/** Remove authorization references in the same transaction as the binding. */
|
|
public static function bindingRemoved(int $id): array
|
|
{
|
|
$garbage = Db::name('mobile_recording_upload')->where('binding_id', $id)->column('upload_id');
|
|
Objects::queueDelete($garbage);
|
|
Db::name('mobile_recording_asset')->where('binding_id', $id)->delete();
|
|
Db::name('mobile_recording_upload')->where('binding_id', $id)->delete();
|
|
return $garbage;
|
|
}
|
|
|
|
public static function summaries(array $ids): array
|
|
{
|
|
if (!$ids) { return []; }
|
|
$result = [];
|
|
foreach ($ids as $id) { $result[$id] = ['audio_available' => false, 'transcript_available' => false, 'upload_state' => 'not_uploaded']; }
|
|
$published = Db::name('mobile_recording_asset')->whereIn('binding_id', $ids)->select()->toArray();
|
|
foreach ($published as $asset) { $result[$asset['binding_id']][$asset['kind'] . '_available'] = true; }
|
|
$pending = Db::name('mobile_recording_upload')->whereIn('binding_id', $ids)->where('state', 'uploading')->column('binding_id');
|
|
foreach ($result as $id => &$summary) {
|
|
$complete = (int) $summary['audio_available'] + (int) $summary['transcript_available'];
|
|
$summary['upload_state'] = in_array($id, array_map('intval', $pending), true) ? 'uploading'
|
|
: ($complete === 2 ? 'complete' : ($complete === 1 ? 'partial' : 'not_uploaded'));
|
|
}
|
|
unset($summary);
|
|
return $result;
|
|
}
|
|
|
|
private static function owned(array $identity, int $diagnosis, int $actor): array
|
|
{
|
|
$info = MobileRecordingLogic::actor($actor, true);
|
|
$binding = Db::name('mobile_recording_binding')->where('admin_id', $actor)
|
|
->where('installation_id', $identity['installation_id'])->where('recording_key', $identity['recording_key'])->lock(true)->find();
|
|
return self::authorizeBinding($binding, $diagnosis, $actor, $info);
|
|
}
|
|
|
|
private static function session(string $id, int $diagnosis, int $actor): array
|
|
{
|
|
$info = MobileRecordingLogic::actor($actor, true);
|
|
$lookup = Db::name('mobile_recording_upload')->where('upload_id', $id)->find();
|
|
$binding = $lookup ? Db::name('mobile_recording_binding')->where('id', $lookup['binding_id'])->where('admin_id', $actor)->lock(true)->find() : null;
|
|
self::authorizeBinding($binding, $diagnosis, $actor, $info);
|
|
$row = Db::name('mobile_recording_upload')->where('upload_id', $id)->lock(true)->find();
|
|
if (!$row || (int) $row['diagnosis_id'] !== $diagnosis) { throw new DomainException('上传会话已失效，请重新准备上传'); }
|
|
return $row;
|
|
}
|
|
|
|
private static function authorizeBinding(?array $binding, int $diagnosis, int $actor, array $info): array
|
|
{
|
|
if (!$binding || (int) $binding['diagnosis_id'] !== $diagnosis) { throw new DomainException('录音关联已变更或不可访问'); }
|
|
MobileRecordingLogic::diagnosis($diagnosis, $actor, $info, true);
|
|
return $binding;
|
|
}
|
|
|
|
private static function synchronize(array $row): array
|
|
{
|
|
if ($row['state'] === 'uploading' && Objects::ready($row)) { return $row; }
|
|
if ($row['state'] === 'uploading') {
|
|
$received = Storage::locked($row['upload_id'], static fn ($file): int => Storage::reconcile($file, (int) $row['received_bytes']));
|
|
if ($received !== (int) $row['received_bytes']) {
|
|
Db::name('mobile_recording_upload')->where('upload_id', $row['upload_id'])->update(['received_bytes' => $received, 'updated_at' => time()]);
|
|
$row['received_bytes'] = $received;
|
|
}
|
|
}
|
|
return $row;
|
|
}
|
|
|
|
private static function mutate(callable $operation): array
|
|
{
|
|
$garbage = [];
|
|
$result = Db::transaction(static function () use ($operation, &$garbage): array { return $operation($garbage); });
|
|
Storage::cleanup($garbage);
|
|
if (isset($result['validation_error'])) { throw new DomainException($result['validation_error']); }
|
|
return $result;
|
|
}
|
|
|
|
private static function deleteUploads(array $ids, array &$garbage): void
|
|
{
|
|
if (!$ids) { return; }
|
|
Objects::queueDelete($ids);
|
|
Db::name('mobile_recording_upload')->whereIn('upload_id', $ids)->delete();
|
|
$garbage = array_merge($garbage, $ids);
|
|
}
|
|
|
|
private static function diagnosisId(array $p): int
|
|
{
|
|
return MobileRecordingLogic::integer($p['diagnosis_id'] ?? null, 1, 4294967295, '诊单标识');
|
|
}
|
|
|
|
private static function uploadId($id): string
|
|
{
|
|
if (!is_string($id) || !preg_match('/^[a-f0-9]{32}$/D', $id)) { throw new DomainException('上传标识无效'); }
|
|
return $id;
|
|
}
|
|
|
|
private static function hash($hash): string
|
|
{
|
|
if (!is_string($hash) || !preg_match('/^[a-f0-9]{64}$/D', $hash)) { throw new DomainException('文件摘要无效'); }
|
|
return $hash;
|
|
}
|
|
|
|
private static function dto(array $row): array
|
|
{
|
|
return ['upload_id' => $row['upload_id'], 'received_bytes' => (int) $row['received_bytes'],
|
|
'size_bytes' => (int) $row['size_bytes'], 'state' => $row['state'], 'chunk_size' => self::CHUNK_SIZE];
|
|
}
|
|
|
|
private static function assetState(array $row): array
|
|
{
|
|
return ['state' => $row['state'], 'upload_id' => $row['upload_id'], 'size_bytes' => (int) $row['size_bytes'],
|
|
'received_bytes' => (int) $row['received_bytes'], 'sha256' => $row['sha256'], 'media_type' => $row['media_type']];
|
|
}
|
|
}
|