diff --git a/apps/api-daemon/src/capture/streamIngest.ts b/apps/api-daemon/src/capture/streamIngest.ts index e919e2b..4db287e 100644 --- a/apps/api-daemon/src/capture/streamIngest.ts +++ b/apps/api-daemon/src/capture/streamIngest.ts @@ -172,7 +172,15 @@ async function runCapture(job: Job): Promise { export function startStreamIngestWorker(): Worker { return new Worker(QUEUE_NAMES.STREAM_INGEST, runCapture, { connection: redisConnection, - concurrency: 3, + // Each capture job runs for the entire duration of a live stream + // (potentially hours) before its BullMQ job resolves. If concurrency is + // lower than the number of channels going live at once, the extra + // channels queue behind already-running captures and may not start + // recording until one of those finishes — i.e. effectively never, for + // streams still live. This must cover the realistic simultaneous-channel + // count; the real ceiling is the server's CPU/bandwidth for running that + // many yt-dlp+ffmpeg pairs at once, not this number. + concurrency: env.maxConcurrentCaptures, // Captures run for the lifetime of the stream (can be hours) — extend the // default lock well beyond BullMQ's stalled-job assumptions for an MVP. lockDuration: 24 * 60 * 60 * 1000, diff --git a/apps/api-daemon/src/env.ts b/apps/api-daemon/src/env.ts index 4d401f0..e0c9cae 100644 --- a/apps/api-daemon/src/env.ts +++ b/apps/api-daemon/src/env.ts @@ -9,4 +9,5 @@ export const env = { forceLiveUrl: process.env.FORCE_LIVE_URL ?? "", ytdlpCookiesB64: process.env.YTDLP_COOKIES_B64 ?? "", ytdlpPotProviderUrl: process.env.YTDLP_POT_PROVIDER_URL ?? "", + maxConcurrentCaptures: Number(process.env.MAX_CONCURRENT_CAPTURES ?? 10), }; diff --git a/apps/frontend/app/actions.ts b/apps/frontend/app/actions.ts index c43d4f0..5929768 100644 --- a/apps/frontend/app/actions.ts +++ b/apps/frontend/app/actions.ts @@ -1,5 +1,6 @@ "use server"; +import { unlink } from "node:fs/promises"; import { prisma } from "@streamclipper/db"; import { revalidatePath } from "next/cache"; @@ -18,3 +19,22 @@ export async function addChannel(formData: FormData) { revalidatePath("/"); } + +export async function deleteRawSegment(segmentId: string) { + const segment = await prisma.rawSegment.findUnique({ where: { id: segmentId } }); + if (!segment) return; + + await prisma.candidateSegment.deleteMany({ where: { rawSegmentId: segmentId } }); + await prisma.rawSegment.delete({ where: { id: segmentId } }); + await unlink(segment.filePath).catch(() => {}); + + revalidatePath("/segments"); + revalidatePath("/", "layout"); +} + +export async function deleteCandidateSegment(candidateId: string) { + await prisma.candidateSegment.delete({ where: { id: candidateId } }); + + revalidatePath("/segments"); + revalidatePath("/", "layout"); +} diff --git a/apps/frontend/app/api/segments/[id]/video/route.ts b/apps/frontend/app/api/segments/[id]/video/route.ts new file mode 100644 index 0000000..7c00eb3 --- /dev/null +++ b/apps/frontend/app/api/segments/[id]/video/route.ts @@ -0,0 +1,48 @@ +import { createReadStream, statSync } from "node:fs"; +import { Readable } from "node:stream"; +import type { NextRequest } from "next/server"; +import { prisma } from "@streamclipper/db"; + +export async function GET(req: NextRequest, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params; + + const segment = await prisma.rawSegment.findUnique({ where: { id } }); + if (!segment) return new Response("Not found", { status: 404 }); + + let size: number; + try { + size = statSync(segment.filePath).size; + } catch { + return new Response("File not found on disk", { status: 404 }); + } + + const headers: Record = { + "Content-Type": "video/mp4", + "Accept-Ranges": "bytes", + }; + if (req.nextUrl.searchParams.get("download") === "1") { + headers["Content-Disposition"] = `attachment; filename="${id}.mp4"`; + } + + const range = req.headers.get("range"); + const match = range ? /bytes=(\d+)-(\d*)/.exec(range) : null; + + if (match) { + const start = Number(match[1]); + const end = match[2] ? Number(match[2]) : size - 1; + const stream = createReadStream(segment.filePath, { start, end }); + return new Response(Readable.toWeb(stream) as ReadableStream, { + status: 206, + headers: { + ...headers, + "Content-Range": `bytes ${start}-${end}/${size}`, + "Content-Length": String(end - start + 1), + }, + }); + } + + const stream = createReadStream(segment.filePath); + return new Response(Readable.toWeb(stream) as ReadableStream, { + headers: { ...headers, "Content-Length": String(size) }, + }); +} diff --git a/apps/frontend/app/channels/[id]/page.tsx b/apps/frontend/app/channels/[id]/page.tsx index a2a00c0..ec298ad 100644 --- a/apps/frontend/app/channels/[id]/page.tsx +++ b/apps/frontend/app/channels/[id]/page.tsx @@ -2,6 +2,8 @@ import { notFound } from "next/navigation"; import Link from "next/link"; import { prisma } from "@streamclipper/db"; import { fetchChannelStatuses } from "../../../lib/apiDaemon"; +import { deleteRawSegment, deleteCandidateSegment } from "../../actions"; +import { DeleteButton } from "../../components/DeleteButton"; export const dynamic = "force-dynamic"; @@ -102,6 +104,17 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{ {segment.status} + + +
+ İndir +
+ + +
+ {segment.candidates.map((c) => (
@@ -114,6 +127,9 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{ {transcriptPreview(c.transcriptJson) && (

{transcriptPreview(c.transcriptJson)}

)} +
+ +
))}
diff --git a/apps/frontend/app/components/DeleteButton.tsx b/apps/frontend/app/components/DeleteButton.tsx new file mode 100644 index 0000000..46b98b4 --- /dev/null +++ b/apps/frontend/app/components/DeleteButton.tsx @@ -0,0 +1,16 @@ +"use client"; + +export function DeleteButton({ confirmText, label = "Sil" }: { confirmText: string; label?: string }) { + return ( + + ); +} diff --git a/apps/frontend/app/segments/page.tsx b/apps/frontend/app/segments/page.tsx index 8df892d..012efa1 100644 --- a/apps/frontend/app/segments/page.tsx +++ b/apps/frontend/app/segments/page.tsx @@ -1,4 +1,6 @@ import { prisma } from "@streamclipper/db"; +import { deleteRawSegment, deleteCandidateSegment } from "../actions"; +import { DeleteButton } from "../components/DeleteButton"; export const dynamic = "force-dynamic"; @@ -40,6 +42,17 @@ export default async function SegmentsPage() { {segment.duration}sn · {new Date(segment.startedAt).toLocaleString("tr-TR")} + + +
+ İndir +
+ + +
+ {segment.candidates.length === 0 ? (

Bu segmentte aday klip bulunamadı. @@ -58,6 +71,9 @@ export default async function SegmentsPage() { {transcriptPreview(c.transcriptJson) && (

{transcriptPreview(c.transcriptJson)}

)} +
+ + )) )} diff --git a/apps/worker/worker/queues.py b/apps/worker/worker/queues.py index 35ef236..b7599ea 100644 --- a/apps/worker/worker/queues.py +++ b/apps/worker/worker/queues.py @@ -16,3 +16,8 @@ QUEUE_NAMES = { REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379") SHARED_MEDIA_ROOT = os.environ.get("SHARED_MEDIA_ROOT", "./shared-media") + +# bullmq-python defaults Worker concurrency to 1, so several channels +# producing segments/candidates around the same time would otherwise process +# strictly one at a time and build up a backlog. +WORKER_CONCURRENCY = int(os.environ.get("WORKER_CONCURRENCY", "5")) diff --git a/apps/worker/worker/signal_detection.py b/apps/worker/worker/signal_detection.py index 03cf358..2d9a60a 100644 --- a/apps/worker/worker/signal_detection.py +++ b/apps/worker/worker/signal_detection.py @@ -11,7 +11,7 @@ import soundfile as sf from bullmq import Queue, Worker from . import db -from .queues import QUEUE_NAMES, REDIS_URL, SHARED_MEDIA_ROOT +from .queues import QUEUE_NAMES, REDIS_URL, SHARED_MEDIA_ROOT, WORKER_CONCURRENCY WINDOW_SEC = 1.0 PEAK_STD_MULTIPLIER = 1.5 @@ -161,4 +161,8 @@ async def process_signal_detection(job, job_token=None): def start_signal_detection_worker() -> Worker: - return Worker(QUEUE_NAMES["SIGNAL_DETECTION"], process_signal_detection, {"connection": REDIS_URL}) + return Worker( + QUEUE_NAMES["SIGNAL_DETECTION"], + process_signal_detection, + {"connection": REDIS_URL, "concurrency": WORKER_CONCURRENCY}, + ) diff --git a/apps/worker/worker/stt_scoring.py b/apps/worker/worker/stt_scoring.py index db37392..c429a5c 100644 --- a/apps/worker/worker/stt_scoring.py +++ b/apps/worker/worker/stt_scoring.py @@ -7,7 +7,7 @@ from bullmq import Worker from openai import OpenAI from . import db -from .queues import QUEUE_NAMES, REDIS_URL +from .queues import QUEUE_NAMES, REDIS_URL, WORKER_CONCURRENCY from .telegram import send_telegram_message _openai_client: OpenAI | None = None @@ -71,4 +71,8 @@ async def process_stt_scoring(job, job_token=None): def start_stt_scoring_worker() -> Worker: - return Worker(QUEUE_NAMES["STT_SCORING"], process_stt_scoring, {"connection": REDIS_URL}) + return Worker( + QUEUE_NAMES["STT_SCORING"], + process_stt_scoring, + {"connection": REDIS_URL, "concurrency": WORKER_CONCURRENCY}, + ) diff --git a/docker-compose.yml b/docker-compose.yml index b08fb73..877d04e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -40,6 +40,7 @@ services: TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-} YTDLP_COOKIES_B64: ${YTDLP_COOKIES_B64:-} YTDLP_POT_PROVIDER_URL: http://sc_pot_provider:4416 + MAX_CONCURRENT_CAPTURES: ${MAX_CONCURRENT_CAPTURES:-10} volumes: - shared-media:/shared-media ports: @@ -62,6 +63,7 @@ services: OPENAI_API_KEY: ${OPENAI_API_KEY:-} TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN:-} TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-} + WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-5} volumes: - shared-media:/shared-media depends_on: @@ -76,6 +78,9 @@ services: environment: DATABASE_URL: postgresql://streamclipper:streamclipper@sc_postgres:5432/streamclipper API_DAEMON_URL: http://sc_api_daemon:4001 + SHARED_MEDIA_ROOT: /shared-media + volumes: + - shared-media:/shared-media ports: - "3000:3000" depends_on: