Spaces:
Running
Running
| ; | |
| var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) { | |
| if (k2 === undefined) k2 = k; | |
| var desc = Object.getOwnPropertyDescriptor(m, k); | |
| if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) { | |
| desc = { enumerable: true, get: function() { return m[k]; } }; | |
| } | |
| Object.defineProperty(o, k2, desc); | |
| }) : (function(o, m, k, k2) { | |
| if (k2 === undefined) k2 = k; | |
| o[k2] = m[k]; | |
| })); | |
| var __setModuleDefault = (this && this.__setModuleDefault) || (Object.create ? (function(o, v) { | |
| Object.defineProperty(o, "default", { enumerable: true, value: v }); | |
| }) : function(o, v) { | |
| o["default"] = v; | |
| }); | |
| var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) { | |
| var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d; | |
| if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc); | |
| else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r; | |
| return c > 3 && r && Object.defineProperty(target, key, r), r; | |
| }; | |
| var __importStar = (this && this.__importStar) || (function () { | |
| var ownKeys = function(o) { | |
| ownKeys = Object.getOwnPropertyNames || function (o) { | |
| var ar = []; | |
| for (var k in o) if (Object.prototype.hasOwnProperty.call(o, k)) ar[ar.length] = k; | |
| return ar; | |
| }; | |
| return ownKeys(o); | |
| }; | |
| return function (mod) { | |
| if (mod && mod.__esModule) return mod; | |
| var result = {}; | |
| if (mod != null) for (var k = ownKeys(mod), i = 0; i < k.length; i++) if (k[i] !== "default") __createBinding(result, mod, k[i]); | |
| __setModuleDefault(result, mod); | |
| return result; | |
| }; | |
| })(); | |
| var __metadata = (this && this.__metadata) || function (k, v) { | |
| if (typeof Reflect === "object" && typeof Reflect.metadata === "function") return Reflect.metadata(k, v); | |
| }; | |
| var __param = (this && this.__param) || function (paramIndex, decorator) { | |
| return function (target, key) { decorator(target, key, paramIndex); } | |
| }; | |
| var ClipScoringProcessor_1; | |
| Object.defineProperty(exports, "__esModule", { value: true }); | |
| exports.ClipScoringProcessor = void 0; | |
| const bullmq_1 = require("@nestjs/bullmq"); | |
| const common_1 = require("@nestjs/common"); | |
| const child_process_1 = require("child_process"); | |
| const util_1 = require("util"); | |
| const path = __importStar(require("path")); | |
| const fs = __importStar(require("fs/promises")); | |
| const os = __importStar(require("os")); | |
| const prisma_service_1 = require("../prisma/prisma.service"); | |
| const r2_storage_service_1 = require("../common/storage/r2-storage.service"); | |
| const redis_constants_1 = require("../common/redis.constants"); | |
| const safe_fetch_1 = require("../common/safe-fetch"); | |
| const execAsync = (0, util_1.promisify)(child_process_1.execFile); | |
| const RELEASE_LOCK_SCRIPT = ` | |
| if redis.call("get", KEYS[1]) == ARGV[1] then | |
| return redis.call("del", KEYS[1]) | |
| else | |
| return 0 | |
| end | |
| `; | |
| const EXTEND_LOCK_SCRIPT = ` | |
| if redis.call("get", KEYS[1]) == ARGV[1] then | |
| return redis.call("pexpire", KEYS[1], ARGV[2]) | |
| else | |
| return 0 | |
| end | |
| `; | |
| let ClipScoringProcessor = ClipScoringProcessor_1 = class ClipScoringProcessor extends bullmq_1.WorkerHost { | |
| prisma; | |
| r2Storage; | |
| logger = new common_1.Logger(ClipScoringProcessor_1.name); | |
| redis; | |
| releaseLockSha = null; | |
| extendLockSha = null; | |
| constructor(prisma, r2Storage, redis) { | |
| super(); | |
| this.prisma = prisma; | |
| this.r2Storage = r2Storage; | |
| this.redis = redis; | |
| } | |
| async onModuleInit() { | |
| try { | |
| this.releaseLockSha = await this.redis.scriptLoad(RELEASE_LOCK_SCRIPT); | |
| this.extendLockSha = await this.redis.scriptLoad(EXTEND_LOCK_SCRIPT); | |
| this.logger.log("Lua scripts compiled for clip-scoring locks"); | |
| } | |
| catch (err) { | |
| this.logger.warn("Failed to pre-compile Lua scripts, will use EVAL fallback"); | |
| } | |
| } | |
| async process(job) { | |
| const { clipId, episodeId, videoUrl, transcriptText } = job.data; | |
| this.logger.log(`Processing clip scoring for ${clipId}`); | |
| const lockKey = `lock:clip:${clipId}`; | |
| const lockValue = `worker:${job.id}:${Date.now()}`; | |
| const lockTTL = 300000; | |
| const acquired = await this.redis.set(lockKey, lockValue, { | |
| NX: true, | |
| PX: lockTTL, | |
| }); | |
| if (!acquired) { | |
| await job.log(`Skipping job ${job.id}: clip ${clipId} is already being scored`); | |
| return { clipId, status: "skipped", reason: "locked" }; | |
| } | |
| const heartbeat = setInterval(async () => { | |
| try { | |
| const extended = await this.evalLua(this.extendLockSha, EXTEND_LOCK_SCRIPT, [lockKey], [lockValue, String(lockTTL)]); | |
| if (!extended) { | |
| this.logger.warn(`Lock lost for clip ${clipId}, aborting heartbeat`); | |
| clearInterval(heartbeat); | |
| } | |
| } | |
| catch (err) { | |
| this.logger.error(`Heartbeat extend failed: ${err}`); | |
| } | |
| }, 60000); | |
| const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), `clip-score-${clipId}-`)); | |
| try { | |
| await job.updateProgress(5); | |
| const sourcePath = path.join(tmpDir, "source.mp4"); | |
| await this.downloadVideo(videoUrl, sourcePath); | |
| await job.updateProgress(15); | |
| const clipBounds = await this.prisma.clip.findUnique({ | |
| where: { id: clipId }, | |
| select: { startSeconds: true, endSeconds: true }, | |
| }); | |
| const clipStart = clipBounds?.startSeconds ?? 0; | |
| const clipEnd = clipBounds?.endSeconds ?? 0; | |
| const clipDuration = Math.max(0, clipEnd - clipStart); | |
| const trimmedPath = path.join(tmpDir, "trimmed.mp4"); | |
| await this.trimSource(sourcePath, trimmedPath, clipStart, clipDuration); | |
| await job.updateProgress(25); | |
| const framesDir = path.join(tmpDir, "frames"); | |
| await fs.mkdir(framesDir, { recursive: true }); | |
| await this.extractFrames(trimmedPath, framesDir); | |
| await job.updateProgress(40); | |
| const audioPath = path.join(tmpDir, "audio.wav"); | |
| await this.extractAudio(trimmedPath, audioPath); | |
| await job.updateProgress(50); | |
| const pythonResult = await this.runPythonPipeline(framesDir, audioPath, transcriptText, tmpDir); | |
| await job.updateProgress(85); | |
| if (pythonResult.status === "error") { | |
| throw new Error(`Python scoring failed: ${pythonResult.message}`); | |
| } | |
| await this.prisma.clip.update({ | |
| where: { id: clipId }, | |
| data: { | |
| visualScore: Math.round(pythonResult.breakdown.visualAesthetic), | |
| audioScore: Math.round(pythonResult.breakdown.audioCompliance), | |
| transcriptScore: Math.round(pythonResult.breakdown.narrativeHook), | |
| scoringMetadata: pythonResult.metadata, | |
| scoredAt: new Date(), | |
| }, | |
| }); | |
| await job.updateProgress(100); | |
| this.logger.log(`Clip ${clipId} scored: overall=${pythonResult.overallScore}, ` + | |
| `visual=${pythonResult.breakdown.visualAesthetic}, ` + | |
| `audio=${pythonResult.breakdown.audioCompliance}, ` + | |
| `hook=${pythonResult.breakdown.narrativeHook}`); | |
| return { | |
| clipId, | |
| status: "completed", | |
| overallScore: pythonResult.overallScore, | |
| breakdown: pythonResult.breakdown, | |
| }; | |
| } | |
| catch (error) { | |
| const msg = error instanceof Error ? error.message : String(error); | |
| this.logger.error(`Clip scoring failed for ${clipId}: ${msg}`); | |
| throw error; | |
| } | |
| finally { | |
| clearInterval(heartbeat); | |
| await this.releaseLock(lockKey, lockValue); | |
| await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => undefined); | |
| } | |
| } | |
| async evalLua(sha, script, keys, args) { | |
| if (sha) { | |
| try { | |
| return await this.redis.evalSha(sha, { keys, arguments: args }); | |
| } | |
| catch (err) { | |
| if (err?.message?.includes("NOSCRIPT")) { | |
| return await this.redis.eval(script, { keys, arguments: args }); | |
| } | |
| throw err; | |
| } | |
| } | |
| return await this.redis.eval(script, { keys, arguments: args }); | |
| } | |
| async releaseLock(lockKey, lockValue) { | |
| try { | |
| await this.evalLua(this.releaseLockSha, RELEASE_LOCK_SCRIPT, [lockKey], [lockValue]); | |
| } | |
| catch (err) { | |
| this.logger.error(`Failed to release lock ${lockKey}: ${err}`); | |
| } | |
| } | |
| async downloadVideo(url, destPath) { | |
| let downloadUrl = url; | |
| if (!url.startsWith("http")) { | |
| downloadUrl = await this.r2Storage.getPresignedUrl(url, 3600); | |
| } | |
| const r2PublicHost = process.env.R2_PUBLIC_HOST || "r2.cloudflarestorage.com"; | |
| const allowedHostSuffixes = [ | |
| `.${r2PublicHost}`, | |
| ".s3.amazonaws.com", | |
| ".storage.googleapis.com", | |
| ".blob.core.windows.net", | |
| ".digitaloceanspaces.com", | |
| ]; | |
| try { | |
| (0, safe_fetch_1.assertSafeUrl)(downloadUrl, { allowedHostSuffixes }); | |
| } | |
| catch (err) { | |
| try { | |
| const u = new URL(downloadUrl); | |
| (0, safe_fetch_1.assertSafeUrl)(downloadUrl, { allowedHosts: [u.hostname] }); | |
| } | |
| catch { | |
| throw err; | |
| } | |
| } | |
| const response = await fetch(downloadUrl); | |
| if (!response.ok) { | |
| throw new Error(`Failed to download video: ${response.status} ${response.statusText}`); | |
| } | |
| const MAX_VIDEO_BYTES = 10 * 1024 * 1024 * 1024; | |
| const contentLength = Number(response.headers.get("content-length")); | |
| if (Number.isFinite(contentLength) && contentLength > MAX_VIDEO_BYTES) { | |
| throw new Error(`Video too large (${contentLength} bytes exceeds the ${MAX_VIDEO_BYTES}-byte cap)`); | |
| } | |
| if (!response.body) { | |
| throw new Error("Downloaded video response has no body stream"); | |
| } | |
| let totalBytes = 0; | |
| const reader = response.body.getReader(); | |
| try { | |
| const fileHandle = await fs.open(destPath, "w"); | |
| try { | |
| for (;;) { | |
| const { done, value } = await reader.read(); | |
| if (done) | |
| break; | |
| totalBytes += value.byteLength; | |
| if (totalBytes > MAX_VIDEO_BYTES) { | |
| throw new Error(`Video exceeds the ${MAX_VIDEO_BYTES}-byte cap; aborting download`); | |
| } | |
| await fileHandle.write(value); | |
| } | |
| } | |
| finally { | |
| await fileHandle.close().catch(() => undefined); | |
| } | |
| } | |
| finally { | |
| reader.cancel().catch(() => undefined); | |
| } | |
| } | |
| async trimSource(sourcePath, outputPath, startSeconds, durationSeconds) { | |
| const args = [ | |
| "-y", | |
| "-ss", String(Math.max(0, startSeconds)), | |
| "-i", sourcePath, | |
| "-t", String(Math.max(0, durationSeconds)), | |
| "-c", "copy", | |
| "-avoid_negative_ts", "make_zero", | |
| outputPath, | |
| ]; | |
| try { | |
| await execAsync("ffmpeg", args, { timeout: 60000 }); | |
| } | |
| catch (err) { | |
| const fallbackArgs = [ | |
| "-y", | |
| "-ss", String(Math.max(0, startSeconds)), | |
| "-i", sourcePath, | |
| "-t", String(Math.max(0, durationSeconds)), | |
| "-c:v", "libx264", | |
| "-c:a", "aac", | |
| "-preset", "ultrafast", | |
| outputPath, | |
| ]; | |
| await execAsync("ffmpeg", fallbackArgs, { timeout: 180000 }); | |
| } | |
| } | |
| async extractFrames(sourcePath, framesDir) { | |
| const outPattern = path.join(framesDir, "frame_%04d.jpg"); | |
| const args = [ | |
| "-y", | |
| "-i", sourcePath, | |
| "-vf", "fps=1,scale=224:224:force_original_aspect_ratio=decrease,pad=224:224:(ow-iw)/2:(oh-ih)/2", | |
| "-q:v", "2", | |
| outPattern, | |
| ]; | |
| try { | |
| await execAsync("ffmpeg", args, { timeout: 120000 }); | |
| } | |
| catch (err) { | |
| const message = err?.message || String(err); | |
| if (message.includes("No such file or directory")) { | |
| throw new Error("FFmpeg not found. Ensure FFmpeg is installed in the worker container."); | |
| } | |
| throw new Error(`FFmpeg frame extraction failed: ${message}`); | |
| } | |
| } | |
| async extractAudio(sourcePath, audioPath) { | |
| const args = [ | |
| "-y", | |
| "-i", sourcePath, | |
| "-vn", | |
| "-acodec", "pcm_s16le", | |
| "-ar", "44100", | |
| audioPath, | |
| ]; | |
| try { | |
| await execAsync("ffmpeg", args, { timeout: 60000 }); | |
| } | |
| catch (err) { | |
| throw new Error(`Audio extraction failed: ${err?.message || String(err)}`); | |
| } | |
| } | |
| async runPythonPipeline(framesDir, audioPath, transcript, tmpDir) { | |
| const possiblePaths = [ | |
| path.resolve(process.cwd(), "services/clip-scorer/pipeline.py"), | |
| path.resolve("/app/scripts/pipeline.py"), | |
| path.resolve(__dirname, "../../scripts/pipeline.py"), | |
| ]; | |
| let scriptPath = ""; | |
| for (const p of possiblePaths) { | |
| try { | |
| await fs.access(p); | |
| scriptPath = p; | |
| break; | |
| } | |
| catch { | |
| continue; | |
| } | |
| } | |
| if (!scriptPath) { | |
| throw new Error("Python pipeline script not found. Searched: " + possiblePaths.join(", ")); | |
| } | |
| const transcriptPath = path.join(tmpDir, "transcript.txt"); | |
| await fs.writeFile(transcriptPath, transcript, "utf8"); | |
| const args = [ | |
| scriptPath, | |
| "--frames_dir", framesDir, | |
| "--audio_path", audioPath, | |
| "--transcript_path", transcriptPath, | |
| ]; | |
| const { stdout, stderr } = await execAsync("python3", args, { | |
| timeout: 300000, | |
| maxBuffer: 1024 * 1024, | |
| }); | |
| if (!stdout.trim()) { | |
| throw new Error(`Python script produced no output. stderr: ${stderr}`); | |
| } | |
| return JSON.parse(stdout.trim()); | |
| } | |
| }; | |
| exports.ClipScoringProcessor = ClipScoringProcessor; | |
| exports.ClipScoringProcessor = ClipScoringProcessor = ClipScoringProcessor_1 = __decorate([ | |
| (0, bullmq_1.Processor)("clip-scoring", { | |
| concurrency: 2, | |
| }), | |
| __param(2, (0, common_1.Inject)(redis_constants_1.REDIS_CLIENT)), | |
| __metadata("design:paramtypes", [prisma_service_1.PrismaService, | |
| r2_storage_service_1.R2StorageService, Object]) | |
| ], ClipScoringProcessor); | |
| //# sourceMappingURL=clip-scoring.processor.js.map |