Feat/media analysis pipeline

backend/alembic/versions/a1b2c3d4e5f6_add_media_analysis_table.py

@@ -0,0 +1,42 @@
+"""Add media_analysis table with pgvector embedding
+
+Revision ID: a1b2c3d4e5f6
+Revises: e8a1b2c3d4f5
+Create Date: 2026-07-28 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = 'a1b2c3d4e5f6'
+down_revision: Union[str, Sequence[str], None] = 'e8a1b2c3d4f5'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+
+    op.create_table(
+        'media_analysis',
+        sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column('video_id', sa.Integer(), sa.ForeignKey('videos.id', ondelete='CASCADE'), nullable=False, unique=True),
+        sa.Column('status', sa.Text(), nullable=False, server_default='pending'),
+        sa.Column('frame_sample_paths', postgresql.JSONB(), nullable=True),
+        sa.Column('scene_cuts', postgresql.JSONB(), nullable=True),
+        sa.Column('beat_timestamps', postgresql.JSONB(), nullable=True),
+        sa.Column('caption_embedding', postgresql.dialects.postgresql.VECTOR(384), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now()),
+    )
+    op.create_index('ix_media_analysis_video_id', 'media_analysis', ['video_id'])
+    op.create_index('ix_media_analysis_status', 'media_analysis', ['status'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_media_analysis_status', table_name='media_analysis')
+    op.drop_index('ix_media_analysis_video_id', table_name='media_analysis')
+    op.drop_table('media_analysis')

backend/alembic/versions/e8a1b2c3d4f5_add_video_events_table.py

@@ -0,0 +1,41 @@
+"""Add video_events table for engagement tracking
+
+Revision ID: e8a1b2c3d4f5
+Revises: 7acd76e1bfde
+Create Date: 2026-07-27 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'e8a1b2c3d4f5'
+down_revision: Union[str, Sequence[str], None] = '7acd76e1bfde'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'video_events',
+        sa.Column('id', sa.BigInteger(), primary_key=True, autoincrement=True),
+        sa.Column('user_id', sa.Integer(), sa.ForeignKey('users.id', ondelete='CASCADE'), nullable=False),
+        sa.Column('video_id', sa.Integer(), sa.ForeignKey('videos.id', ondelete='CASCADE'), nullable=False),
+        sa.Column('event_type', sa.Text(), nullable=False),
+        sa.Column('watch_seconds', sa.Integer(), nullable=True),
+        sa.Column('session_id', sa.Text(), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
+    )
+    op.create_index('ix_video_events_user_created', 'video_events', ['user_id', 'created_at'])
+    op.create_index('ix_video_events_video_event', 'video_events', ['video_id', 'event_type'])
+    op.create_index('ix_video_events_video_created', 'video_events', ['video_id', 'created_at'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_video_events_video_created', table_name='video_events')
+    op.drop_index('ix_video_events_video_event', table_name='video_events')
+    op.drop_index('ix_video_events_user_created', table_name='video_events')
+    op.drop_table('video_events')

backend/app/api/v1/api.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter
-from app.api.v1.endpoints import auth, videos, admin, users, posts, achievements, notifications, ads, chat, challenges, monetization, video_views, seo, recommendations, partners, categories, library, history, watch_later, liked, following, metrics, email, reports, comments
+from app.api.v1.endpoints import auth, videos, admin, users, posts, achievements, notifications, ads, chat, challenges, monetization, video_views, seo, recommendations, partners, categories, library, history, watch_later, liked, following, metrics, email, reports, comments, events
 
 
 api_router = APIRouter()
@@ -28,6 +28,7 @@
 api_router.include_router(email.router, prefix="/email", tags=["email"])
 api_router.include_router(reports.router, prefix="/reports", tags=["reports"])
 api_router.include_router(comments.router, prefix="/comments", tags=["comments"])
+api_router.include_router(events.router, prefix="/events", tags=["events"])
 
 
 

backend/app/api/v1/endpoints/events.py

@@ -0,0 +1,39 @@
+import json
+import logging
+from typing import List, Union
+
+from fastapi import APIRouter, Response
+from app.schemas.video_events import VideoEventIn, VideoEventBatch
+from app.core.redis import redis_client
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+PENDING_KEY = "video_events:pending"
+
+
+@router.post("/video", status_code=202)
+def ingest_video_events(payload: Union[VideoEventBatch, List[VideoEventIn]]):
+    """Accept one or a batch of video engagement events and push to Redis for async processing."""
+    events = payload.events if isinstance(payload, VideoEventBatch) else payload
+
+    if not events:
+        return Response(status_code=202)
+
+    pipe = redis_client.pipeline()
+    for ev in events:
+        pipe.rpush(
+            PENDING_KEY,
+            json.dumps({
+                "user_id": ev.user_id,
+                "video_id": ev.video_id,
+                "event_type": ev.event_type.value,
+                "watch_seconds": ev.watch_seconds,
+                "session_id": ev.session_id,
+            }),
+        )
+    pipe.execute()
+
+    logger.debug("Enqueued %d video events to Redis", len(events))
+    return Response(status_code=202)

backend/app/api/v1/endpoints/videos.py

@@ -1510,3 +1510,20 @@ def delete_comment(
     if result is False:
         raise HTTPException(status_code=403, detail="Not authorized to delete this comment")
     return {"status": "success", "message": "Comment deleted successfully"}
+
+
+# ── Media Analysis ────────────────────────────────────────────────────────────
+
+@router.get("/{video_id}/analysis")
+def get_media_analysis(video_id: str, db: Session = Depends(get_db)):
+    """Return the media_analysis row for a video (frames, scene cuts, beats, embedding)."""
+    from sqlalchemy import text
+    row = db.execute(
+        text("SELECT * FROM media_analysis WHERE video_id = :vid"),
+        {"vid": video_id},
+    ).mappings().first()
+
+    if not row:
+        raise HTTPException(status_code=404, detail="No analysis found for this video")
+
+    return dict(row)

backend/app/schemas/video_events.py

@@ -0,0 +1,36 @@
+from pydantic import BaseModel, field_validator
+from typing import Optional, List, Union
+import enum
+
+
+class VideoEventType(str, enum.Enum):
+    VIEW = "view"
+    WATCH_25 = "watch_25"
+    WATCH_50 = "watch_50"
+    WATCH_75 = "watch_75"
+    COMPLETE = "complete"
+    LIKE = "like"
+    SHARE = "share"
+    SKIP = "skip"
+
+
+class VideoEventIn(BaseModel):
+    user_id: int
+    video_id: int
+    event_type: VideoEventType
+    watch_seconds: Optional[int] = None
+    session_id: Optional[str] = None
+
+    @field_validator("event_type", mode="before")
+    @classmethod
+    def coerce_event_type(cls, v):
+        if isinstance(v, VideoEventType):
+            return v
+        try:
+            return VideoEventType(v)
+        except ValueError:
+            raise ValueError(f"Invalid event_type: {v}. Must be one of: {[e.value for e in VideoEventType]}")
+
+
+class VideoEventBatch(BaseModel):
+    events: List[VideoEventIn]

backend/app/services/media_analysis.py

@@ -0,0 +1,133 @@
+"""
+Media analysis pipeline — extracts frames, scene cuts, beat timestamps,
+and caption embeddings from a transcoded video.
+
+Called by the Celery task after transcoding completes.
+"""
+import os
+import tempfile
+import shutil
+import logging
+import subprocess
+import json
+from typing import List, Optional
+
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+def extract_frames(video_path: str, video_id: str) -> List[str]:
+    """Sample 1 frame per second from the video, upload each to S3, return S3 keys."""
+    from app.core.storage import storage
+
+    frames_dir = tempfile.mkdtemp(prefix="frames_")
+    try:
+        subprocess.run(
+            [
+                "ffmpeg", "-i", video_path,
+                "-vf", "fps=1",
+                "-q:v", "3",
+                os.path.join(frames_dir, "frame_%04d.jpg"),
+            ],
+            check=True,
+            capture_output=True,
+            timeout=600,
+        )
+
+        s3_keys = []
+        for fname in sorted(os.listdir(frames_dir)):
+            local_path = os.path.join(frames_dir, fname)
+            s3_key = f"analysis/{video_id}/frames/{fname}"
+            storage.upload_file(local_path, s3_key)
+            s3_keys.append(s3_key)
+
+        logger.info("Uploaded %d frames for video %s", len(s3_keys), video_id)
+        return s3_keys
+    finally:
+        shutil.rmtree(frames_dir, ignore_errors=True)
+
+
+def detect_scene_cuts(video_path: str) -> List[float]:
+    """Detect scene cuts using pyscenedetect ContentDetector. Returns timestamps in seconds."""
+    from scenedetect import open_video, SceneManager
+    from scenedetect.detectors import ContentDetector
+
+    video = open_video(video_path)
+    scene_manager = SceneManager()
+    scene_manager.add_detector(ContentDetector(threshold=27.0))
+    scene_manager.detect_scenes(video, show_progress=False)
+
+    scene_list = scene_manager.get_scene_list()
+    # Return the start time of each scene as seconds
+    timestamps = [round(scene[0].get_seconds(), 3) for scene in scene_list]
+    logger.info("Detected %d scene cuts", len(timestamps))
+    return timestamps
+
+
+def detect_beats(video_path: str) -> List[float]:
+    """Extract audio and detect beat timestamps using librosa. Returns timestamps in seconds."""
+    import librosa
+
+    # Extract audio to a temp WAV file
+    audio_tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+    audio_tmp.close()
+    try:
+        subprocess.run(
+            [
+                "ffmpeg", "-i", video_path,
+                "-vn", "-acodec", "pcm_s16le", "-ar", "22050", "-ac", "1",
+                "-y", audio_tmp.name,
+            ],
+            check=True,
+            capture_output=True,
+            timeout=300,
+        )
+
+        y, sr = librosa.load(audio_tmp.name, sr=22050)
+        tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
+        beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist()
+        beat_times = [round(t, 3) for t in beat_times]
+        logger.info("Detected %d beat timestamps", len(beat_times))
+        return beat_times
+    finally:
+        os.unlink(audio_tmp.name)
+
+
+def generate_caption_embedding(title: str, tags: Optional[str]) -> Optional[List[float]]:
+    """Generate a 384-dim embedding from title+tags using all-MiniLM-L6-v2."""
+    from sentence_transformers import SentenceTransformer
+
+    text = title or ""
+    if tags:
+        tag_str = " ".join(t.strip() for t in tags.split(",") if t.strip())
+        text = f"{text} {tag_str}"
+
+    text = text.strip()
+    if not text:
+        return None
+
+    model = SentenceTransformer("all-MiniLM-L6-v2")
+    embedding = model.encode(text, normalize_embeddings=True)
+    return embedding.tolist()
+
+
+def download_video_from_s3(video_s3_key: str) -> str:
+    """Download a video from S3 to a temp path. Returns the local file path."""
+    from app.core.storage import storage
+    from app.core import config
+
+    suffix = os.path.splitext(video_s3_key)[-1] or ".mp4"
+    tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
+    tmp.close()
+
+    try:
+        storage.s3_client.download_file(
+            config.AWS_STORAGE_BUCKET_NAME,
+            video_s3_key,
+            tmp.name,
+        )
+        return tmp.name
+    except Exception:
+        os.unlink(tmp.name)
+        raise

backend/app/tasks/event_tasks.py

@@ -0,0 +1,55 @@
+import json
+import logging
+from datetime import datetime, timezone
+
+from celery import shared_task
+from sqlalchemy import text
+
+# Bind @shared_task to the Redis-backed worker app
+import app.worker  # noqa: F401
+from app.db.session import SessionLocal
+from app.core.redis import redis_client
+
+logger = logging.getLogger(__name__)
+
+PENDING_KEY = "video_events:pending"
+FAILED_KEY = "video_events:failed"
+BATCH_SIZE = 500
+
+
+@shared_task(name="tasks.events.flush_video_events")
+def flush_video_events():
+    """LPOP up to 500 pending events from Redis and bulk-insert into video_events."""
+    events = []
+    for _ in range(BATCH_SIZE):
+        raw = redis_client.lpop(PENDING_KEY)
+        if raw is None:
+            break
+        try:
+            events.append(json.loads(raw))
+        except (json.JSONDecodeError, TypeError):
+            logger.warning("Skipping malformed event: %s", raw)
+
+    if not events:
+        return
+
+    db = SessionLocal()
+    try:
+        db.execute(
+            text(
+                "INSERT INTO video_events (user_id, video_id, event_type, watch_seconds, session_id, created_at) "
+                "VALUES (:user_id, :video_id, :event_type, :watch_seconds, :session_id, now())"
+            ),
+            events,
+        )
+        db.commit()
+        logger.info("Flushed %d video events to database", len(events))
+    except Exception as e:
+        db.rollback()
+        logger.error("Failed to flush %d video events: %s", len(events), e)
+        pipe = redis_client.pipeline()
+        for ev in events:
+            pipe.rpush(FAILED_KEY, json.dumps(ev))
+        pipe.execute()
+    finally:
+        db.close()