StreamClipper AI Faz 1: ingestion + sinyal analizi + STT + altyapı iskeleti
yt-dlp headless capture (15dk segmentleme), ses peak + chat velocity sinyal tespiti, OpenAI Whisper STT (kelime zaman damgalı), BullMQ/Redis/Postgres altyapısı ve kanal durumu + transkript kütüphanesi gösteren Next.js panel. LLM virality skorlama, render/crop/altyazı ve multi-platform dağıtım bu fazın kapsamı dışında. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,13 @@
|
||||
FROM python:3.12-slim
|
||||
|
||||
RUN apt-get update && apt-get install -y --no-install-recommends \
|
||||
ffmpeg libsndfile1 \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
WORKDIR /app
|
||||
COPY apps/worker/requirements.txt ./apps/worker/requirements.txt
|
||||
RUN pip install --no-cache-dir -r apps/worker/requirements.txt
|
||||
COPY apps/worker ./apps/worker
|
||||
|
||||
WORKDIR /app/apps/worker
|
||||
CMD ["python", "-m", "worker.main"]
|
||||
@@ -0,0 +1,8 @@
|
||||
bullmq>=2.9.0
|
||||
asyncpg>=0.29.0
|
||||
numpy>=1.26.0
|
||||
soundfile>=0.12.1
|
||||
openai>=1.50.0
|
||||
httpx>=0.27.0
|
||||
chat-downloader>=0.2.7
|
||||
python-dotenv>=1.0.0
|
||||
@@ -0,0 +1,48 @@
|
||||
"""
|
||||
Captures a YouTube live chat as timestamped JSONL for the lifetime of a
|
||||
StreamSession. chat-downloader's `get_chat` is a blocking generator, so each
|
||||
session's capture runs on its own daemon thread.
|
||||
"""
|
||||
|
||||
import json
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from chat_downloader import ChatDownloader
|
||||
|
||||
from .queues import SHARED_MEDIA_ROOT
|
||||
|
||||
_active_sessions: set[str] = set()
|
||||
|
||||
|
||||
def _capture_loop(session_id: str, video_id: str) -> None:
|
||||
out_dir = Path(SHARED_MEDIA_ROOT) / "raw" / session_id
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
chat_path = out_dir / "chat.jsonl"
|
||||
url = f"https://www.youtube.com/watch?v={video_id}"
|
||||
|
||||
print(f"[chat-capture] started for session {session_id} ({url})")
|
||||
try:
|
||||
chat = ChatDownloader().get_chat(url)
|
||||
with chat_path.open("a", encoding="utf-8") as f:
|
||||
for message in chat:
|
||||
record = {
|
||||
"timestamp": message.get("timestamp"),
|
||||
"time_in_seconds": message.get("time_in_seconds"),
|
||||
"author": (message.get("author") or {}).get("name"),
|
||||
"message": message.get("message"),
|
||||
}
|
||||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||||
f.flush()
|
||||
except Exception as exc: # chat-downloader raises when the stream ends
|
||||
print(f"[chat-capture] session {session_id} stopped: {exc}")
|
||||
finally:
|
||||
_active_sessions.discard(session_id)
|
||||
|
||||
|
||||
def start_chat_capture(session_id: str, video_id: str | None) -> None:
|
||||
if session_id in _active_sessions or not video_id or video_id == "forced":
|
||||
return
|
||||
_active_sessions.add(session_id)
|
||||
thread = threading.Thread(target=_capture_loop, args=(session_id, video_id), daemon=True)
|
||||
thread.start()
|
||||
@@ -0,0 +1,95 @@
|
||||
"""
|
||||
Raw asyncpg access against the same Postgres tables Prisma manages from the
|
||||
Node side (packages/db/prisma/schema.prisma). Prisma's `@default(cuid())` is
|
||||
applied by Prisma Client at insert time, not as a Postgres-level DEFAULT, so
|
||||
rows written from here generate their own ids (uuid4) instead.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
|
||||
import asyncpg
|
||||
|
||||
_pool: asyncpg.Pool | None = None
|
||||
|
||||
|
||||
async def get_pool() -> asyncpg.Pool:
|
||||
global _pool
|
||||
if _pool is None:
|
||||
_pool = await asyncpg.create_pool(dsn=os.environ["DATABASE_URL"])
|
||||
return _pool
|
||||
|
||||
|
||||
async def get_raw_segment(raw_segment_id: str) -> asyncpg.Record | None:
|
||||
pool = await get_pool()
|
||||
return await pool.fetchrow(
|
||||
"""
|
||||
SELECT rs.id, rs.file_path, rs.duration, rs.session_id
|
||||
FROM raw_segments rs
|
||||
WHERE rs.id = $1
|
||||
""",
|
||||
raw_segment_id,
|
||||
)
|
||||
|
||||
|
||||
async def insert_candidate_segment(
|
||||
raw_segment_id: str,
|
||||
start_sec: int,
|
||||
end_sec: int,
|
||||
audio_peak_score: float,
|
||||
chat_velocity_score: float | None,
|
||||
) -> str:
|
||||
pool = await get_pool()
|
||||
candidate_id = str(uuid.uuid4())
|
||||
await pool.execute(
|
||||
"""
|
||||
INSERT INTO candidate_segments
|
||||
(id, raw_segment_id, start_sec, end_sec, audio_peak_score, chat_velocity_score, status, created_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, 'PENDING_STT', now())
|
||||
""",
|
||||
candidate_id,
|
||||
raw_segment_id,
|
||||
start_sec,
|
||||
end_sec,
|
||||
audio_peak_score,
|
||||
chat_velocity_score,
|
||||
)
|
||||
return candidate_id
|
||||
|
||||
|
||||
async def mark_raw_segment_processed(raw_segment_id: str) -> None:
|
||||
pool = await get_pool()
|
||||
await pool.execute(
|
||||
"UPDATE raw_segments SET status = 'PROCESSED' WHERE id = $1", raw_segment_id
|
||||
)
|
||||
|
||||
|
||||
async def get_candidate_segment(candidate_id: str) -> asyncpg.Record | None:
|
||||
pool = await get_pool()
|
||||
return await pool.fetchrow(
|
||||
"""
|
||||
SELECT cs.id, cs.start_sec, cs.end_sec, rs.file_path, c.name AS channel_name
|
||||
FROM candidate_segments cs
|
||||
JOIN raw_segments rs ON rs.id = cs.raw_segment_id
|
||||
JOIN stream_sessions ss ON ss.id = rs.session_id
|
||||
JOIN channels c ON c.id = ss.channel_id
|
||||
WHERE cs.id = $1
|
||||
""",
|
||||
candidate_id,
|
||||
)
|
||||
|
||||
|
||||
async def update_candidate_transcript(candidate_id: str, transcript: dict, status: str) -> None:
|
||||
pool = await get_pool()
|
||||
await pool.execute(
|
||||
"UPDATE candidate_segments SET transcript_json = $2::jsonb, status = $3 WHERE id = $1",
|
||||
candidate_id,
|
||||
json.dumps(transcript),
|
||||
status,
|
||||
)
|
||||
|
||||
|
||||
async def get_active_sessions() -> list[asyncpg.Record]:
|
||||
pool = await get_pool()
|
||||
return await pool.fetch("SELECT id, live_video_id FROM stream_sessions WHERE ended_at IS NULL")
|
||||
@@ -0,0 +1,35 @@
|
||||
import asyncio
|
||||
|
||||
from . import db
|
||||
from .chat_capture import start_chat_capture
|
||||
from .signal_detection import start_signal_detection_worker
|
||||
from .stt_scoring import start_stt_scoring_worker
|
||||
|
||||
SESSION_WATCH_INTERVAL_SEC = 10
|
||||
|
||||
|
||||
async def watch_sessions_for_chat_capture():
|
||||
while True:
|
||||
try:
|
||||
sessions = await db.get_active_sessions()
|
||||
for s in sessions:
|
||||
start_chat_capture(s["id"], s["live_video_id"])
|
||||
except Exception as exc:
|
||||
print(f"[main] session watcher error: {exc}")
|
||||
await asyncio.sleep(SESSION_WATCH_INTERVAL_SEC)
|
||||
|
||||
|
||||
async def main():
|
||||
signal_worker = start_signal_detection_worker()
|
||||
stt_worker = start_stt_scoring_worker()
|
||||
print("[main] worker started: signal-detection + stt-scoring queues")
|
||||
|
||||
try:
|
||||
await watch_sessions_for_chat_capture()
|
||||
finally:
|
||||
await signal_worker.close()
|
||||
await stt_worker.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,18 @@
|
||||
import os
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
# Kept in sync with apps/api-daemon/src/queues.ts — the protocol-compatible
|
||||
# `bullmq` package on both sides lets Node and Python share these Redis queues.
|
||||
QUEUE_NAMES = {
|
||||
"STREAM_INGEST": "stream-ingest",
|
||||
"SIGNAL_DETECTION": "signal-detection",
|
||||
"STT_SCORING": "stt-scoring",
|
||||
"VIDEO_RENDER": "video-render",
|
||||
"POST_PUBLISH": "post-publish",
|
||||
}
|
||||
|
||||
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
|
||||
SHARED_MEDIA_ROOT = os.environ.get("SHARED_MEDIA_ROOT", "./shared-media")
|
||||
@@ -0,0 +1,164 @@
|
||||
import asyncio
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import soundfile as sf
|
||||
from bullmq import Queue, Worker
|
||||
|
||||
from . import db
|
||||
from .queues import QUEUE_NAMES, REDIS_URL, SHARED_MEDIA_ROOT
|
||||
|
||||
WINDOW_SEC = 1.0
|
||||
PEAK_STD_MULTIPLIER = 1.5
|
||||
CANDIDATE_PAD_SEC = 45
|
||||
CHAT_SPIKE_MULTIPLIER = 3
|
||||
SEGMENT_TIME_SEC = int(os.environ.get("SEGMENT_TIME_SEC", "900"))
|
||||
|
||||
stt_scoring_queue = Queue(QUEUE_NAMES["STT_SCORING"], {"connection": REDIS_URL})
|
||||
|
||||
|
||||
def _extract_audio_peaks(file_path: str) -> list[tuple[float, float]]:
|
||||
"""(time_sec, db_level) for 1s windows whose RMS dB is a statistical outlier."""
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav") as tmp:
|
||||
subprocess.run(
|
||||
["ffmpeg", "-y", "-i", file_path, "-ac", "1", "-ar", "16000", "-vn", tmp.name],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
data, sr = sf.read(tmp.name)
|
||||
|
||||
if data.ndim > 1:
|
||||
data = data.mean(axis=1)
|
||||
|
||||
window_samples = int(WINDOW_SEC * sr)
|
||||
n_windows = max(1, len(data) // window_samples) if window_samples else 0
|
||||
db_levels = []
|
||||
for i in range(n_windows):
|
||||
chunk = data[i * window_samples : (i + 1) * window_samples]
|
||||
rms = float(np.sqrt(np.mean(chunk.astype(np.float64) ** 2))) if len(chunk) else 0.0
|
||||
db_levels.append(20 * math.log10(rms + 1e-9))
|
||||
|
||||
if not db_levels:
|
||||
return []
|
||||
|
||||
mean_db = float(np.mean(db_levels))
|
||||
std_db = float(np.std(db_levels)) or 1.0
|
||||
threshold = mean_db + PEAK_STD_MULTIPLIER * std_db
|
||||
|
||||
return [(i * WINDOW_SEC, level) for i, level in enumerate(db_levels) if level > threshold]
|
||||
|
||||
|
||||
def _load_chat_velocity_peaks(session_id: str) -> list[float]:
|
||||
"""Absolute (stream-relative) second offsets of minutes with a chat spike."""
|
||||
chat_path = Path(SHARED_MEDIA_ROOT) / "raw" / session_id / "chat.jsonl"
|
||||
if not chat_path.exists():
|
||||
return []
|
||||
|
||||
buckets: dict[int, int] = {}
|
||||
with chat_path.open(encoding="utf-8") as f:
|
||||
for line in f:
|
||||
try:
|
||||
record = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
t = record.get("time_in_seconds")
|
||||
if t is None:
|
||||
continue
|
||||
minute = int(t // 60)
|
||||
buckets[minute] = buckets.get(minute, 0) + 1
|
||||
|
||||
if not buckets:
|
||||
return []
|
||||
|
||||
avg = sum(buckets.values()) / len(buckets)
|
||||
return [minute * 60.0 for minute, count in buckets.items() if count > CHAT_SPIKE_MULTIPLIER * avg]
|
||||
|
||||
|
||||
def _segment_index_from_filename(file_path: str) -> int:
|
||||
stem = Path(file_path).stem # segment_003
|
||||
try:
|
||||
return int(stem.rsplit("_", 1)[-1])
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
|
||||
def _build_candidate_windows(
|
||||
audio_peaks: list[tuple[float, float]],
|
||||
chat_peak_times_abs: list[float],
|
||||
segment_abs_offset: float,
|
||||
segment_duration: int,
|
||||
) -> list[tuple[int, int, float, float | None]]:
|
||||
windows: list[tuple[int, int, float, float | None]] = []
|
||||
|
||||
for peak_time, peak_db in audio_peaks:
|
||||
abs_time = segment_abs_offset + peak_time
|
||||
nearest_chat = min(
|
||||
(c for c in chat_peak_times_abs if abs(c - abs_time) <= 60),
|
||||
key=lambda c: abs(c - abs_time),
|
||||
default=None,
|
||||
)
|
||||
center = (abs_time + nearest_chat) / 2 if nearest_chat is not None else abs_time
|
||||
|
||||
# Known simplification: candidate windows are clamped to the current
|
||||
# 15-minute segment file since STT later cuts audio from that single
|
||||
# file. A peak near a segment boundary can get truncated — stitching
|
||||
# across segments is a future improvement, not part of this phase.
|
||||
rel_start = max(0, int(center - segment_abs_offset - CANDIDATE_PAD_SEC))
|
||||
rel_end = min(segment_duration, int(center - segment_abs_offset + CANDIDATE_PAD_SEC))
|
||||
if rel_end - rel_start < 5:
|
||||
continue
|
||||
|
||||
chat_score = 1.0 if nearest_chat is not None else None
|
||||
windows.append((rel_start, rel_end, peak_db, chat_score))
|
||||
|
||||
windows.sort(key=lambda w: w[0])
|
||||
merged: list[list] = []
|
||||
for w in windows:
|
||||
if merged and w[0] <= merged[-1][1]:
|
||||
merged[-1][1] = max(merged[-1][1], w[1])
|
||||
merged[-1][2] = max(merged[-1][2], w[2])
|
||||
else:
|
||||
merged.append(list(w))
|
||||
|
||||
return [tuple(m) for m in merged]
|
||||
|
||||
|
||||
async def process_signal_detection(job, job_token=None):
|
||||
raw_segment_id = job.data["rawSegmentId"]
|
||||
file_path = job.data["filePath"]
|
||||
session_id = job.data["sessionId"]
|
||||
|
||||
print(f"[signal-detection] analyzing {file_path}")
|
||||
|
||||
raw_segment = await db.get_raw_segment(raw_segment_id)
|
||||
if raw_segment is None:
|
||||
print(f"[signal-detection] raw_segment {raw_segment_id} not found, skipping")
|
||||
return
|
||||
|
||||
audio_peaks = await asyncio.to_thread(_extract_audio_peaks, file_path)
|
||||
chat_peaks_abs = await asyncio.to_thread(_load_chat_velocity_peaks, session_id)
|
||||
|
||||
segment_index = _segment_index_from_filename(file_path)
|
||||
segment_abs_offset = segment_index * SEGMENT_TIME_SEC
|
||||
|
||||
windows = _build_candidate_windows(
|
||||
audio_peaks, chat_peaks_abs, segment_abs_offset, raw_segment["duration"]
|
||||
)
|
||||
|
||||
for start_sec, end_sec, audio_score, chat_score in windows:
|
||||
candidate_id = await db.insert_candidate_segment(
|
||||
raw_segment_id, start_sec, end_sec, audio_score, chat_score
|
||||
)
|
||||
await stt_scoring_queue.add("score-candidate", {"candidateSegmentId": candidate_id})
|
||||
print(f"[signal-detection] candidate {candidate_id} [{start_sec}-{end_sec}]s")
|
||||
|
||||
await db.mark_raw_segment_processed(raw_segment_id)
|
||||
|
||||
|
||||
def start_signal_detection_worker() -> Worker:
|
||||
return Worker(QUEUE_NAMES["SIGNAL_DETECTION"], process_signal_detection, {"connection": REDIS_URL})
|
||||
@@ -0,0 +1,74 @@
|
||||
import asyncio
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
from bullmq import Worker
|
||||
from openai import OpenAI
|
||||
|
||||
from . import db
|
||||
from .queues import QUEUE_NAMES, REDIS_URL
|
||||
from .telegram import send_telegram_message
|
||||
|
||||
_openai_client: OpenAI | None = None
|
||||
|
||||
|
||||
def _client() -> OpenAI:
|
||||
global _openai_client
|
||||
if _openai_client is None:
|
||||
_openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
|
||||
return _openai_client
|
||||
|
||||
|
||||
def _transcribe(clip_path: str) -> dict:
|
||||
with open(clip_path, "rb") as f:
|
||||
transcript = _client().audio.transcriptions.create(
|
||||
model="whisper-1",
|
||||
file=f,
|
||||
response_format="verbose_json",
|
||||
timestamp_granularities=["word"],
|
||||
)
|
||||
return transcript.model_dump()
|
||||
|
||||
|
||||
async def process_stt_scoring(job, job_token=None):
|
||||
candidate_id = job.data["candidateSegmentId"]
|
||||
candidate = await db.get_candidate_segment(candidate_id)
|
||||
if candidate is None:
|
||||
print(f"[stt-scoring] candidate {candidate_id} not found, skipping")
|
||||
return
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".mp3") as clip:
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg", "-y",
|
||||
"-ss", str(candidate["start_sec"]),
|
||||
"-to", str(candidate["end_sec"]),
|
||||
"-i", candidate["file_path"],
|
||||
"-ac", "1", "-ar", "16000",
|
||||
clip.name,
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
try:
|
||||
transcript_dict = await asyncio.to_thread(_transcribe, clip.name)
|
||||
except Exception as exc:
|
||||
print(f"[stt-scoring] OpenAI call failed for {candidate_id}: {exc}")
|
||||
await db.update_candidate_transcript(candidate_id, {"error": str(exc)}, "FAILED")
|
||||
return
|
||||
|
||||
await db.update_candidate_transcript(candidate_id, transcript_dict, "TRANSCRIBED")
|
||||
|
||||
preview = (transcript_dict.get("text") or "")[:200]
|
||||
await send_telegram_message(
|
||||
f"🎬 *{candidate['channel_name']}* için yeni aday klip transkribe edildi "
|
||||
f"({candidate['start_sec']}-{candidate['end_sec']}s):\n_{preview}_"
|
||||
)
|
||||
|
||||
print(f"[stt-scoring] candidate {candidate_id} transcribed")
|
||||
|
||||
|
||||
def start_stt_scoring_worker() -> Worker:
|
||||
return Worker(QUEUE_NAMES["STT_SCORING"], process_stt_scoring, {"connection": REDIS_URL})
|
||||
@@ -0,0 +1,20 @@
|
||||
import os
|
||||
|
||||
import httpx
|
||||
|
||||
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
|
||||
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
|
||||
|
||||
|
||||
async def send_telegram_message(text: str) -> None:
|
||||
if not BOT_TOKEN or not CHAT_ID:
|
||||
print(f"[telegram] not configured, skipping notification: {text}")
|
||||
return
|
||||
|
||||
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
url, json={"chat_id": CHAT_ID, "text": text, "parse_mode": "Markdown"}
|
||||
)
|
||||
if resp.status_code >= 400:
|
||||
print(f"[telegram] failed to send message: {resp.status_code} {resp.text}")
|
||||
Reference in New Issue
Block a user