Fix stuck transcriptions and make transcripts stoppable and searchable.
Raise queue retry_after above the job timeout, recover orphaned runs, allow stop/restart with any engine, and keep finished transcripts searchable in the recordings list.
This commit is contained in:
@@ -2,9 +2,17 @@
|
||||
|
||||
namespace App\Models;
|
||||
|
||||
use App\Jobs\TranscribeRecording;
|
||||
use Carbon\Carbon;
|
||||
use Illuminate\Database\Eloquent\Attributes\Scope;
|
||||
use Illuminate\Database\Eloquent\Builder;
|
||||
use Illuminate\Database\Eloquent\Casts\Attribute;
|
||||
use Illuminate\Database\Eloquent\Model;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Facades\Storage;
|
||||
use Illuminate\Support\Str;
|
||||
use Throwable;
|
||||
|
||||
class Recording extends Model
|
||||
{
|
||||
@@ -86,6 +94,233 @@ class Recording extends Model
|
||||
return in_array($this->transcription_status, ['pending', 'processing'], true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Search title, metadata, and stored transcript text.
|
||||
*/
|
||||
#[Scope]
|
||||
protected function search(Builder $query, string $term): void
|
||||
{
|
||||
$like = '%'.$term.'%';
|
||||
|
||||
$query->where(function (Builder $builder) use ($like): void {
|
||||
$builder->where('title', 'like', $like)
|
||||
->orWhere('artist', 'like', $like)
|
||||
->orWhere('album', 'like', $like)
|
||||
->orWhere('original_filename', 'like', $like)
|
||||
->orWhere('transcript', 'like', $like);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Short transcript excerpt, optionally centered on a search hit.
|
||||
*/
|
||||
public function transcriptSnippet(?string $term = null, int $radius = 80): ?string
|
||||
{
|
||||
if (! filled($this->transcript)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$transcript = preg_replace('/\s+/', ' ', $this->transcript) ?? $this->transcript;
|
||||
|
||||
if ($term === null || $term === '') {
|
||||
return Str::limit($transcript, $radius * 2);
|
||||
}
|
||||
|
||||
$position = mb_stripos($transcript, $term);
|
||||
|
||||
if ($position === false) {
|
||||
return Str::limit($transcript, $radius * 2);
|
||||
}
|
||||
|
||||
$start = max(0, $position - $radius);
|
||||
$excerpt = mb_substr($transcript, $start, ($radius * 2) + mb_strlen($term));
|
||||
|
||||
return ($start > 0 ? '…' : '').$excerpt.(mb_strlen($transcript) > $start + mb_strlen($excerpt) ? '…' : '');
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a TranscribeRecording job for this recording is still on the queue.
|
||||
*/
|
||||
public function hasActiveTranscriptionJob(): bool
|
||||
{
|
||||
return DB::table('jobs')
|
||||
->pluck('payload')
|
||||
->contains(function (string $payload): bool {
|
||||
if (! str_contains($payload, TranscribeRecording::class)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$data = json_decode($payload, true);
|
||||
$command = $data['data']['command'] ?? null;
|
||||
|
||||
if (! is_string($command)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
$job = unserialize($command);
|
||||
} catch (Throwable) {
|
||||
return (bool) preg_match('/id";i:'.$this->id.';/', $payload);
|
||||
}
|
||||
|
||||
return $job instanceof TranscribeRecording
|
||||
&& (int) $job->recording->getKey() === (int) $this->id;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Processing/pending with no worker job left (crashed worker, bad retry_after, etc.).
|
||||
*/
|
||||
public function isOrphanedTranscription(): bool
|
||||
{
|
||||
if (! $this->isTranscribing()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if ($this->hasActiveTranscriptionJob()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$reference = $this->transcription_started_at ?? $this->updated_at;
|
||||
|
||||
// Allow a short window after dispatch before the row appears / worker claims it.
|
||||
if ($reference !== null && $reference->gt(now()->subSeconds(15))) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether this recording still expects results for the given run.
|
||||
*/
|
||||
public function ownsTranscriptionRun(?string $runStartedAt): bool
|
||||
{
|
||||
$this->refresh();
|
||||
|
||||
if (! $this->isTranscribing()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if ($runStartedAt === null || $this->transcription_started_at === null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return $this->transcription_started_at->getTimestamp() === Carbon::parse($runStartedAt)->getTimestamp();
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove queued TranscribeRecording jobs for this recording.
|
||||
*/
|
||||
public function discardQueuedTranscriptionJobs(): int
|
||||
{
|
||||
$deleted = 0;
|
||||
|
||||
DB::table('jobs')
|
||||
->orderBy('id')
|
||||
->get()
|
||||
->each(function (object $job) use (&$deleted): void {
|
||||
$payload = (string) $job->payload;
|
||||
|
||||
if (! str_contains($payload, TranscribeRecording::class)) {
|
||||
return;
|
||||
}
|
||||
|
||||
$data = json_decode($payload, true);
|
||||
$command = $data['data']['command'] ?? null;
|
||||
|
||||
if (! is_string($command)) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
$queued = unserialize($command);
|
||||
} catch (Throwable) {
|
||||
if (! preg_match('/id";i:'.$this->id.';/', $payload)) {
|
||||
return;
|
||||
}
|
||||
|
||||
DB::table('jobs')->where('id', $job->id)->delete();
|
||||
$deleted++;
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
if ($queued instanceof TranscribeRecording && (int) $queued->recording->getKey() === (int) $this->id) {
|
||||
DB::table('jobs')->where('id', $job->id)->delete();
|
||||
$deleted++;
|
||||
}
|
||||
});
|
||||
|
||||
$this->releaseTranscriptionUniqueLock();
|
||||
|
||||
return $deleted;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop transcription: drop queued jobs and mark the run cancelled.
|
||||
*
|
||||
* @param bool $silent When true, skip status update (used before starting a replacement run).
|
||||
*/
|
||||
public function cancelTranscription(bool $silent = false): void
|
||||
{
|
||||
$this->discardQueuedTranscriptionJobs();
|
||||
|
||||
if ($silent) {
|
||||
return;
|
||||
}
|
||||
|
||||
$this->forceFill([
|
||||
'transcription_status' => 'cancelled',
|
||||
'transcription_progress' => 'Stopped by user',
|
||||
'transcription_percent' => $this->transcription_percent ?: 0,
|
||||
'transcription_error' => 'Stopped by user',
|
||||
])->save();
|
||||
}
|
||||
|
||||
/**
|
||||
* Release a leftover ShouldBeUnique lock from earlier job versions.
|
||||
*/
|
||||
public function releaseTranscriptionUniqueLock(): void
|
||||
{
|
||||
Cache::lock(
|
||||
'laravel_unique_job:'.TranscribeRecording::class.'transcribe-recording:'.$this->id
|
||||
)->forceRelease();
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark transcription as failed and unblock the UI.
|
||||
*/
|
||||
public function markTranscriptionFailed(string $message): void
|
||||
{
|
||||
if (! $this->isTranscribing()) {
|
||||
return;
|
||||
}
|
||||
|
||||
$this->forceFill([
|
||||
'transcription_status' => 'failed',
|
||||
'transcription_progress' => 'Transcription failed',
|
||||
'transcription_percent' => $this->transcription_percent ?: 0,
|
||||
'transcription_error' => $message,
|
||||
])->save();
|
||||
}
|
||||
|
||||
/**
|
||||
* Recover a stuck transcription if the queue job is gone.
|
||||
*/
|
||||
public function recoverOrphanedTranscription(): bool
|
||||
{
|
||||
if (! $this->isOrphanedTranscription()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$this->markTranscriptionFailed(
|
||||
'Transcription worker stopped before finishing. Start transcription again.',
|
||||
);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the live progress fields shown in the UI.
|
||||
*/
|
||||
@@ -141,6 +376,7 @@ class Recording extends Model
|
||||
'duration_seconds' => $this->duration_seconds,
|
||||
'is_active' => $this->isTranscribing(),
|
||||
'has_transcript' => filled($this->transcript),
|
||||
'transcript' => $this->transcript,
|
||||
'transcribed_at' => $this->transcribed_at?->toIso8601String(),
|
||||
];
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user