feat: panelden video izleme/indirme/silme + eşzamanlı yayın kapasitesi düzeltmesi
- streamIngest worker concurrency 3'ten sabit değildi, artık MAX_CONCURRENT_CAPTURES (varsayılan 10) ile ayarlanabiliyor. Eskiden 3'ten fazla kanal aynı anda canlıya geçerse fazlası, ilk 3'ten biri bitene kadar (saatlerce) hiç kayda alınmıyordu. - Python worker'larda (signal-detection, stt-scoring) bullmq varsayılan concurrency'si 1'di — WORKER_CONCURRENCY (varsayılan 5) ile paralelleştirildi, birden fazla kanal aynı anda segment/transkript üretirse sıraya takılmasın diye. - Frontend artık shared-media volume'üne bağlı: /api/segments/[id]/video route'u Range destekli video stream/indirme sağlıyor. Segment ve kanal detay sayfalarına <video> oynatıcı, indirme linki ve onaylı silme (RawSegment + CandidateSegment, dosya dahil) eklendi. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -172,7 +172,15 @@ async function runCapture(job: Job<StreamIngestJob>): Promise<void> {
|
|||||||
export function startStreamIngestWorker(): Worker<StreamIngestJob> {
|
export function startStreamIngestWorker(): Worker<StreamIngestJob> {
|
||||||
return new Worker<StreamIngestJob>(QUEUE_NAMES.STREAM_INGEST, runCapture, {
|
return new Worker<StreamIngestJob>(QUEUE_NAMES.STREAM_INGEST, runCapture, {
|
||||||
connection: redisConnection,
|
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
|
// 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.
|
// default lock well beyond BullMQ's stalled-job assumptions for an MVP.
|
||||||
lockDuration: 24 * 60 * 60 * 1000,
|
lockDuration: 24 * 60 * 60 * 1000,
|
||||||
|
|||||||
@@ -9,4 +9,5 @@ export const env = {
|
|||||||
forceLiveUrl: process.env.FORCE_LIVE_URL ?? "",
|
forceLiveUrl: process.env.FORCE_LIVE_URL ?? "",
|
||||||
ytdlpCookiesB64: process.env.YTDLP_COOKIES_B64 ?? "",
|
ytdlpCookiesB64: process.env.YTDLP_COOKIES_B64 ?? "",
|
||||||
ytdlpPotProviderUrl: process.env.YTDLP_POT_PROVIDER_URL ?? "",
|
ytdlpPotProviderUrl: process.env.YTDLP_POT_PROVIDER_URL ?? "",
|
||||||
|
maxConcurrentCaptures: Number(process.env.MAX_CONCURRENT_CAPTURES ?? 10),
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
"use server";
|
"use server";
|
||||||
|
|
||||||
|
import { unlink } from "node:fs/promises";
|
||||||
import { prisma } from "@streamclipper/db";
|
import { prisma } from "@streamclipper/db";
|
||||||
import { revalidatePath } from "next/cache";
|
import { revalidatePath } from "next/cache";
|
||||||
|
|
||||||
@@ -18,3 +19,22 @@ export async function addChannel(formData: FormData) {
|
|||||||
|
|
||||||
revalidatePath("/");
|
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");
|
||||||
|
}
|
||||||
|
|||||||
@@ -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<string, string> = {
|
||||||
|
"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) },
|
||||||
|
});
|
||||||
|
}
|
||||||
@@ -2,6 +2,8 @@ import { notFound } from "next/navigation";
|
|||||||
import Link from "next/link";
|
import Link from "next/link";
|
||||||
import { prisma } from "@streamclipper/db";
|
import { prisma } from "@streamclipper/db";
|
||||||
import { fetchChannelStatuses } from "../../../lib/apiDaemon";
|
import { fetchChannelStatuses } from "../../../lib/apiDaemon";
|
||||||
|
import { deleteRawSegment, deleteCandidateSegment } from "../../actions";
|
||||||
|
import { DeleteButton } from "../../components/DeleteButton";
|
||||||
|
|
||||||
export const dynamic = "force-dynamic";
|
export const dynamic = "force-dynamic";
|
||||||
|
|
||||||
@@ -102,6 +104,17 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{
|
|||||||
<span className={`badge ${STATUS_BADGE[segment.status] ?? "muted"}`}>{segment.status}</span>
|
<span className={`badge ${STATUS_BADGE[segment.status] ?? "muted"}`}>{segment.status}</span>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<video controls preload="metadata" style={{ width: "100%", maxWidth: "480px", marginTop: "0.5rem", borderRadius: "6px" }}>
|
||||||
|
<source src={`/api/segments/${segment.id}/video`} type="video/mp4" />
|
||||||
|
</video>
|
||||||
|
|
||||||
|
<div style={{ display: "flex", gap: "0.5rem", alignItems: "center", marginTop: "0.5rem" }}>
|
||||||
|
<a className="badge muted" href={`/api/segments/${segment.id}/video?download=1`}>İndir</a>
|
||||||
|
<form action={deleteRawSegment.bind(null, segment.id)}>
|
||||||
|
<DeleteButton confirmText="Bu segmenti ve içindeki tüm aday klipleri silmek istediğine emin misin? Geri alınamaz." />
|
||||||
|
</form>
|
||||||
|
</div>
|
||||||
|
|
||||||
{segment.candidates.map((c) => (
|
{segment.candidates.map((c) => (
|
||||||
<div key={c.id} style={{ marginTop: "0.5rem", paddingLeft: "0.75rem", borderLeft: "2px solid var(--border)" }}>
|
<div key={c.id} style={{ marginTop: "0.5rem", paddingLeft: "0.75rem", borderLeft: "2px solid var(--border)" }}>
|
||||||
<div className="card-head">
|
<div className="card-head">
|
||||||
@@ -114,6 +127,9 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{
|
|||||||
{transcriptPreview(c.transcriptJson) && (
|
{transcriptPreview(c.transcriptJson) && (
|
||||||
<p className="transcript-preview">{transcriptPreview(c.transcriptJson)}</p>
|
<p className="transcript-preview">{transcriptPreview(c.transcriptJson)}</p>
|
||||||
)}
|
)}
|
||||||
|
<form action={deleteCandidateSegment.bind(null, c.id)} style={{ marginTop: "0.35rem" }}>
|
||||||
|
<DeleteButton confirmText="Bu aday klibi silmek istediğine emin misin?" />
|
||||||
|
</form>
|
||||||
</div>
|
</div>
|
||||||
))}
|
))}
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
"use client";
|
||||||
|
|
||||||
|
export function DeleteButton({ confirmText, label = "Sil" }: { confirmText: string; label?: string }) {
|
||||||
|
return (
|
||||||
|
<button
|
||||||
|
type="submit"
|
||||||
|
className="badge err"
|
||||||
|
style={{ border: "none", cursor: "pointer" }}
|
||||||
|
onClick={(e) => {
|
||||||
|
if (!confirm(confirmText)) e.preventDefault();
|
||||||
|
}}
|
||||||
|
>
|
||||||
|
{label}
|
||||||
|
</button>
|
||||||
|
);
|
||||||
|
}
|
||||||
@@ -1,4 +1,6 @@
|
|||||||
import { prisma } from "@streamclipper/db";
|
import { prisma } from "@streamclipper/db";
|
||||||
|
import { deleteRawSegment, deleteCandidateSegment } from "../actions";
|
||||||
|
import { DeleteButton } from "../components/DeleteButton";
|
||||||
|
|
||||||
export const dynamic = "force-dynamic";
|
export const dynamic = "force-dynamic";
|
||||||
|
|
||||||
@@ -40,6 +42,17 @@ export default async function SegmentsPage() {
|
|||||||
{segment.duration}sn · {new Date(segment.startedAt).toLocaleString("tr-TR")}
|
{segment.duration}sn · {new Date(segment.startedAt).toLocaleString("tr-TR")}
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<video controls preload="metadata" style={{ width: "100%", maxWidth: "480px", marginTop: "0.5rem", borderRadius: "6px" }}>
|
||||||
|
<source src={`/api/segments/${segment.id}/video`} type="video/mp4" />
|
||||||
|
</video>
|
||||||
|
|
||||||
|
<div style={{ display: "flex", gap: "0.5rem", alignItems: "center", marginTop: "0.5rem" }}>
|
||||||
|
<a className="badge muted" href={`/api/segments/${segment.id}/video?download=1`}>İndir</a>
|
||||||
|
<form action={deleteRawSegment.bind(null, segment.id)}>
|
||||||
|
<DeleteButton confirmText="Bu segmenti ve içindeki tüm aday klipleri silmek istediğine emin misin? Geri alınamaz." />
|
||||||
|
</form>
|
||||||
|
</div>
|
||||||
|
|
||||||
{segment.candidates.length === 0 ? (
|
{segment.candidates.length === 0 ? (
|
||||||
<p className="empty" style={{ marginTop: "0.5rem" }}>
|
<p className="empty" style={{ marginTop: "0.5rem" }}>
|
||||||
Bu segmentte aday klip bulunamadı.
|
Bu segmentte aday klip bulunamadı.
|
||||||
@@ -58,6 +71,9 @@ export default async function SegmentsPage() {
|
|||||||
{transcriptPreview(c.transcriptJson) && (
|
{transcriptPreview(c.transcriptJson) && (
|
||||||
<p className="transcript-preview">{transcriptPreview(c.transcriptJson)}</p>
|
<p className="transcript-preview">{transcriptPreview(c.transcriptJson)}</p>
|
||||||
)}
|
)}
|
||||||
|
<form action={deleteCandidateSegment.bind(null, c.id)} style={{ marginTop: "0.35rem" }}>
|
||||||
|
<DeleteButton confirmText="Bu aday klibi silmek istediğine emin misin?" />
|
||||||
|
</form>
|
||||||
</div>
|
</div>
|
||||||
))
|
))
|
||||||
)}
|
)}
|
||||||
|
|||||||
@@ -16,3 +16,8 @@ QUEUE_NAMES = {
|
|||||||
|
|
||||||
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
|
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
|
||||||
SHARED_MEDIA_ROOT = os.environ.get("SHARED_MEDIA_ROOT", "./shared-media")
|
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"))
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import soundfile as sf
|
|||||||
from bullmq import Queue, Worker
|
from bullmq import Queue, Worker
|
||||||
|
|
||||||
from . import db
|
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
|
WINDOW_SEC = 1.0
|
||||||
PEAK_STD_MULTIPLIER = 1.5
|
PEAK_STD_MULTIPLIER = 1.5
|
||||||
@@ -161,4 +161,8 @@ async def process_signal_detection(job, job_token=None):
|
|||||||
|
|
||||||
|
|
||||||
def start_signal_detection_worker() -> Worker:
|
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},
|
||||||
|
)
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ from bullmq import Worker
|
|||||||
from openai import OpenAI
|
from openai import OpenAI
|
||||||
|
|
||||||
from . import db
|
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
|
from .telegram import send_telegram_message
|
||||||
|
|
||||||
_openai_client: OpenAI | None = None
|
_openai_client: OpenAI | None = None
|
||||||
@@ -71,4 +71,8 @@ async def process_stt_scoring(job, job_token=None):
|
|||||||
|
|
||||||
|
|
||||||
def start_stt_scoring_worker() -> Worker:
|
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},
|
||||||
|
)
|
||||||
|
|||||||
@@ -40,6 +40,7 @@ services:
|
|||||||
TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-}
|
TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-}
|
||||||
YTDLP_COOKIES_B64: ${YTDLP_COOKIES_B64:-}
|
YTDLP_COOKIES_B64: ${YTDLP_COOKIES_B64:-}
|
||||||
YTDLP_POT_PROVIDER_URL: http://sc_pot_provider:4416
|
YTDLP_POT_PROVIDER_URL: http://sc_pot_provider:4416
|
||||||
|
MAX_CONCURRENT_CAPTURES: ${MAX_CONCURRENT_CAPTURES:-10}
|
||||||
volumes:
|
volumes:
|
||||||
- shared-media:/shared-media
|
- shared-media:/shared-media
|
||||||
ports:
|
ports:
|
||||||
@@ -62,6 +63,7 @@ services:
|
|||||||
OPENAI_API_KEY: ${OPENAI_API_KEY:-}
|
OPENAI_API_KEY: ${OPENAI_API_KEY:-}
|
||||||
TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN:-}
|
TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN:-}
|
||||||
TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-}
|
TELEGRAM_CHAT_ID: ${TELEGRAM_CHAT_ID:-}
|
||||||
|
WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-5}
|
||||||
volumes:
|
volumes:
|
||||||
- shared-media:/shared-media
|
- shared-media:/shared-media
|
||||||
depends_on:
|
depends_on:
|
||||||
@@ -76,6 +78,9 @@ services:
|
|||||||
environment:
|
environment:
|
||||||
DATABASE_URL: postgresql://streamclipper:streamclipper@sc_postgres:5432/streamclipper
|
DATABASE_URL: postgresql://streamclipper:streamclipper@sc_postgres:5432/streamclipper
|
||||||
API_DAEMON_URL: http://sc_api_daemon:4001
|
API_DAEMON_URL: http://sc_api_daemon:4001
|
||||||
|
SHARED_MEDIA_ROOT: /shared-media
|
||||||
|
volumes:
|
||||||
|
- shared-media:/shared-media
|
||||||
ports:
|
ports:
|
||||||
- "3000:3000"
|
- "3000:3000"
|
||||||
depends_on:
|
depends_on:
|
||||||
|
|||||||
Reference in New Issue
Block a user