fix: NoSuchKey, The specified key does not exist.
This commit is contained in:
@@ -339,9 +339,9 @@ export async function encodeAndUploadHls(opts: EncodeAndUploadOptions): Promise<
|
|||||||
emit({ stage: 'uploading', percent: 73 });
|
emit({ stage: 'uploading', percent: 73 });
|
||||||
await uploader.flush();
|
await uploader.flush();
|
||||||
|
|
||||||
// HLS audio segments are enough for server-side Whisper prep — no second
|
// Prefer the dedicated HLS audio playlist when present; otherwise fall
|
||||||
// full pass through the source file for a separate audio.m4a upload.
|
// back to the highest uploaded video rendition (muxed audio is fine for Whisper).
|
||||||
const audioKey = audioTrack ? `${assetId}/audio/playlist.m3u8` : undefined;
|
const audioKey = audioTrack ? resolveTranscriptionAudioKey(assetId, rungs, uploader) : undefined;
|
||||||
|
|
||||||
// --- Thumbnails ----------------------------------------------------
|
// --- Thumbnails ----------------------------------------------------
|
||||||
emit({ stage: 'thumbnail', percent: 86 });
|
emit({ stage: 'thumbnail', percent: 86 });
|
||||||
@@ -744,6 +744,7 @@ class HlsUploader {
|
|||||||
private uploadedCount = 0;
|
private uploadedCount = 0;
|
||||||
private inflight: Promise<void>[] = [];
|
private inflight: Promise<void>[] = [];
|
||||||
private uploadFailures: unknown[] = [];
|
private uploadFailures: unknown[] = [];
|
||||||
|
private readonly uploadedPaths = new Set<string>();
|
||||||
private readonly uploadGate = new ConcurrencyGate(HLS_UPLOAD_CONCURRENCY);
|
private readonly uploadGate = new ConcurrencyGate(HLS_UPLOAD_CONCURRENCY);
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
@@ -758,6 +759,10 @@ class HlsUploader {
|
|||||||
}
|
}
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
|
hasUploaded(relativePath: string): boolean {
|
||||||
|
return this.uploadedPaths.has(relativePath);
|
||||||
|
}
|
||||||
|
|
||||||
targetFor(path: string, mimeType: string): BufferTarget {
|
targetFor(path: string, mimeType: string): BufferTarget {
|
||||||
return new BufferTarget({
|
return new BufferTarget({
|
||||||
onFinalize: (buffer) => {
|
onFinalize: (buffer) => {
|
||||||
@@ -801,6 +806,7 @@ class HlsUploader {
|
|||||||
});
|
});
|
||||||
this.options.presignedUrls.set(path, presigned);
|
this.options.presignedUrls.set(path, presigned);
|
||||||
|
|
||||||
|
this.uploadedPaths.add(path);
|
||||||
this.totalBytes += u8.byteLength;
|
this.totalBytes += u8.byteLength;
|
||||||
this.uploadedCount += 1;
|
this.uploadedCount += 1;
|
||||||
this.options.onUploaded({ totalBytes: this.totalBytes, count: this.uploadedCount });
|
this.options.onUploaded({ totalBytes: this.totalBytes, count: this.uploadedCount });
|
||||||
@@ -960,6 +966,28 @@ function pad(n: number, width: number): string {
|
|||||||
return n.toString().padStart(width, '0');
|
return n.toString().padStart(width, '0');
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function resolveTranscriptionAudioKey(
|
||||||
|
assetId: string,
|
||||||
|
rungs: HlsRungConfig[],
|
||||||
|
uploader: HlsUploader
|
||||||
|
): string | undefined {
|
||||||
|
if (uploader.hasUploaded('audio/playlist.m3u8')) {
|
||||||
|
return `${assetId}/audio/playlist.m3u8`;
|
||||||
|
}
|
||||||
|
|
||||||
|
for (let index = rungs.length - 1; index >= 0; index -= 1) {
|
||||||
|
const rung = rungs[index];
|
||||||
|
if (!rung) continue;
|
||||||
|
|
||||||
|
const playlistPath = `${rung.name}/playlist.m3u8`;
|
||||||
|
if (uploader.hasUploaded(playlistPath)) {
|
||||||
|
return `${assetId}/${playlistPath}`;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
/** Rough byte budget for progress UI — actual HLS output varies with content complexity. */
|
/** Rough byte budget for progress UI — actual HLS output varies with content complexity. */
|
||||||
function estimateHlsUploadBytes(rungs: HlsRungConfig[], durationSeconds: number, hasAudio: boolean): number {
|
function estimateHlsUploadBytes(rungs: HlsRungConfig[], durationSeconds: number, hasAudio: boolean): number {
|
||||||
const segmentCount = Math.max(1, Math.ceil(durationSeconds / HLS_TARGET_SEGMENT_SECONDS)) + 5;
|
const segmentCount = Math.max(1, Math.ceil(durationSeconds / HLS_TARGET_SEGMENT_SECONDS)) + 5;
|
||||||
|
|||||||
@@ -2,12 +2,13 @@ import { mkdir, readFile, rename, rm, stat, writeFile } from 'node:fs/promises';
|
|||||||
import path from 'node:path';
|
import path from 'node:path';
|
||||||
import { tmpdir } from 'node:os';
|
import { tmpdir } from 'node:os';
|
||||||
|
|
||||||
import { getJobStep, updateMediaJob, upsertJobStep } from '@cio/db/queries';
|
import { getAssetById, getJobStep, updateMediaJob, upsertJobStep } from '@cio/db/queries';
|
||||||
import { ZExtractAudioPayload, type TExtractAudioPayload } from '@cio/jobs/payloads/media';
|
import { ZExtractAudioPayload, type TExtractAudioPayload } from '@cio/jobs/payloads/media';
|
||||||
|
|
||||||
import {
|
import {
|
||||||
downloadObjectToTempFile,
|
downloadObjectToTempFile,
|
||||||
mediaBucket,
|
mediaBucket,
|
||||||
|
objectExists,
|
||||||
safeUnlink,
|
safeUnlink,
|
||||||
uploadFileToBucket,
|
uploadFileToBucket,
|
||||||
videosBucket
|
videosBucket
|
||||||
@@ -83,6 +84,60 @@ async function downloadHlsPlaylistToTempFile(
|
|||||||
return { inputPath, tempDir };
|
return { inputPath, tempDir };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type HlsAssetMetadata = {
|
||||||
|
hlsRenditions?: string[];
|
||||||
|
};
|
||||||
|
|
||||||
|
function readHlsRenditions(metadata: unknown): string[] {
|
||||||
|
if (typeof metadata !== 'object' || metadata === null || Array.isArray(metadata)) {
|
||||||
|
return [];
|
||||||
|
}
|
||||||
|
|
||||||
|
return (metadata as HlsAssetMetadata).hlsRenditions ?? [];
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Pick the first videos-bucket object that actually exists. HLS assets may
|
||||||
|
* store a dedicated audio playlist, a legacy audio.m4a, or only muxed audio
|
||||||
|
* inside a video rendition playlist.
|
||||||
|
*/
|
||||||
|
async function resolveTranscriptionSourceKey(assetId: string, payloadStorageKey: string): Promise<string> {
|
||||||
|
const asset = await getAssetById(assetId);
|
||||||
|
if (!asset) {
|
||||||
|
throw new Error(`Asset ${assetId} not found`);
|
||||||
|
}
|
||||||
|
|
||||||
|
const candidates: string[] = [];
|
||||||
|
const push = (key: string | null | undefined) => {
|
||||||
|
if (key && !candidates.includes(key)) {
|
||||||
|
candidates.push(key);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
push(asset.hlsAudioKey);
|
||||||
|
push(payloadStorageKey);
|
||||||
|
push(asset.storageKey);
|
||||||
|
|
||||||
|
if (asset.hlsManifestKey) {
|
||||||
|
const prefix = `${assetId}/`;
|
||||||
|
push(`${prefix}audio/playlist.m3u8`);
|
||||||
|
push(`${prefix}audio.m4a`);
|
||||||
|
for (const rung of [...readHlsRenditions(asset.metadata)].reverse()) {
|
||||||
|
push(`${prefix}${rung}/playlist.m3u8`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const bucket = videosBucket();
|
||||||
|
for (const key of candidates) {
|
||||||
|
if (await objectExists(bucket, key)) {
|
||||||
|
log.info('extract-audio-source-resolved', { assetId, key });
|
||||||
|
return key;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
throw new Error(`No transcription source object found for asset ${assetId} (tried ${candidates.join(', ')})`);
|
||||||
|
}
|
||||||
|
|
||||||
async function resolveAudioSourcePath(
|
async function resolveAudioSourcePath(
|
||||||
bucket: string,
|
bucket: string,
|
||||||
storageKey: string
|
storageKey: string
|
||||||
@@ -125,7 +180,8 @@ export async function processExtractAudio(rawData: unknown): Promise<ExtractAudi
|
|||||||
let inputTempDir: string | undefined;
|
let inputTempDir: string | undefined;
|
||||||
let outputPath: string | undefined;
|
let outputPath: string | undefined;
|
||||||
try {
|
try {
|
||||||
const source = await resolveAudioSourcePath(videosBucket(), storageKey);
|
const resolvedStorageKey = await resolveTranscriptionSourceKey(assetId, storageKey);
|
||||||
|
const source = await resolveAudioSourcePath(videosBucket(), resolvedStorageKey);
|
||||||
inputPath = source.inputPath;
|
inputPath = source.inputPath;
|
||||||
inputTempDir = source.tempDir;
|
inputTempDir = source.tempDir;
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { GetObjectCommand, PutObjectCommand } from '@aws-sdk/client-s3';
|
import { GetObjectCommand, HeadObjectCommand, NotFound, PutObjectCommand } from '@aws-sdk/client-s3';
|
||||||
import { Readable } from 'node:stream';
|
import { Readable } from 'node:stream';
|
||||||
import { createWriteStream } from 'node:fs';
|
import { createWriteStream } from 'node:fs';
|
||||||
import { mkdir, readFile, unlink } from 'node:fs/promises';
|
import { mkdir, readFile, unlink } from 'node:fs/promises';
|
||||||
@@ -13,6 +13,24 @@ import { getS3Client, getStorageConfig } from '../config/storage';
|
|||||||
* Workers prefer disk over buffers so ffmpeg can seek without holding the
|
* Workers prefer disk over buffers so ffmpeg can seek without holding the
|
||||||
* whole video in RAM.
|
* whole video in RAM.
|
||||||
*/
|
*/
|
||||||
|
export async function objectExists(bucket: string, key: string): Promise<boolean> {
|
||||||
|
try {
|
||||||
|
await getS3Client().send(new HeadObjectCommand({ Bucket: bucket, Key: key }));
|
||||||
|
return true;
|
||||||
|
} catch (error) {
|
||||||
|
if (error instanceof NotFound || (error as { name?: string }).name === 'NotFound') {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
const status = (error as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode;
|
||||||
|
if (status === 404) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
export async function downloadObjectToTempFile(bucket: string, key: string, fileNameHint?: string): Promise<string> {
|
export async function downloadObjectToTempFile(bucket: string, key: string, fileNameHint?: string): Promise<string> {
|
||||||
const dir = path.join(tmpdir(), 'cio-jobs');
|
const dir = path.join(tmpdir(), 'cio-jobs');
|
||||||
await mkdir(dir, { recursive: true });
|
await mkdir(dir, { recursive: true });
|
||||||
|
|||||||
@@ -1209,10 +1209,9 @@ export async function finalizeHls1080Service(orgId: string, assetId: string, dat
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* HLS assets ship audio as a separate `audio.m4a` already extracted by
|
* HLS assets ship audio as HLS segments (dedicated `audio/` playlist or muxed
|
||||||
* Mediabunny in the browser. The existing transcription pipeline accepts
|
* into a video rendition). The transcription pipeline demuxes those segments
|
||||||
* any media key in the videos bucket — its extract-audio leg is a no-op
|
* server-side before calling Whisper.
|
||||||
* on input that's already audio.
|
|
||||||
*/
|
*/
|
||||||
async function enqueueAudioTranscriptionForAsset(input: {
|
async function enqueueAudioTranscriptionForAsset(input: {
|
||||||
organizationId: string;
|
organizationId: string;
|
||||||
|
|||||||
Reference in New Issue
Block a user