refactor: keep dependancies maintainable

api/main.py

@@ -7,6 +7,7 @@
     python -m uvicorn api.main:app --host 0.0.0.0 --port 8000
 """
 
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
@@ -19,7 +20,7 @@
 
 
 @asynccontextmanager
-async def lifespan(app: FastAPI):  # type: ignore
+async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     """Startup and shutdown events."""
     # Startup
     logger.info("Starting Streaming Video-RAG API...")
@@ -69,7 +70,11 @@ def health() -> dict[str, str]:
     return {
         "status": "ok",
         "llm_provider": settings.llm_provider.value,
-        "llm_model": settings.llm_model if settings.llm_provider.value != "ollama" else settings.ollama_model,
+        "llm_model": (
+            settings.llm_model
+            if settings.llm_provider.value != "ollama"
+            else settings.ollama_model
+        ),
         "whisper_mode": settings.whisper_mode.value,
         "whisper_model": settings.whisper_model_size,
         "embedding_mode": settings.embedding_mode.value,

api/models.py

@@ -8,14 +8,23 @@
 
 
 class IngestRequest(BaseModel):
-    source: str = Field(..., description="YouTube URL, local file path, stream URL, or video API URL")
+    source: str = Field(
+        ..., description="YouTube URL, local file path, stream URL, or video API URL"
+    )
     source_type: str | None = Field(
         None,
         description="Hint for source type: 'youtube' | 'local_file' | 'live_stream' | 'video_api'. Auto-detected if omitted.",
     )
-    language: str | None = Field(None, description="ISO 639-1 language code (e.g. 'en'). Auto-detected if omitted.")
-    platform: str | None = Field(None, description="For video_api source: 'vimeo' | 'twitch'")
-    api_credentials: dict[str, str] | None = Field(None, description="API credentials for video_api source")
+    language: str | None = Field(
+        None,
+        description="ISO 639-1 language code (e.g. 'en'). Auto-detected if omitted.",
+    )
+    platform: str | None = Field(
+        None, description="For video_api source: 'vimeo' | 'twitch'"
+    )
+    api_credentials: dict[str, str] | None = Field(
+        None, description="API credentials for video_api source"
+    )
 
 
 class IngestResponse(BaseModel):
@@ -59,9 +68,15 @@ class VideoListResponse(BaseModel):
 
 
 class QueryRequest(BaseModel):
-    question: str = Field(..., description="Natural language question to answer from video content")
-    video_id: str | None = Field(None, description="If set, restrict search to this video")
-    top_k: int | None = Field(None, description="Number of chunks to retrieve (default: from settings)")
+    question: str = Field(
+        ..., description="Natural language question to answer from video content"
+    )
+    video_id: str | None = Field(
+        None, description="If set, restrict search to this video"
+    )
+    top_k: int | None = Field(
+        None, description="Number of chunks to retrieve (default: from settings)"
+    )
 
 
 class SourceCitation(BaseModel):
@@ -86,7 +101,9 @@ class QueryResponse(BaseModel):
 
 class SearchRequest(BaseModel):
     query: str = Field(..., description="Semantic search query")
-    video_id: str | None = Field(None, description="If set, search only within this video")
+    video_id: str | None = Field(
+        None, description="If set, search only within this video"
+    )
     top_k: int = Field(10, description="Maximum number of results")
     min_score: float = Field(0.0, description="Minimum similarity score (0-1)")
 
@@ -117,7 +134,9 @@ class SearchResponse(BaseModel):
 
 class SummarizeRequest(BaseModel):
     video_id: str
-    include_chapters: bool = Field(True, description="Also generate per-chapter summaries if available")
+    include_chapters: bool = Field(
+        True, description="Also generate per-chapter summaries if available"
+    )
 
 
 class ChapterSummary(BaseModel):

api/routes/ingest.py

@@ -144,7 +144,9 @@ def _run_ingest_pipeline(
 
     db = SessionLocal()
 
-    def update_job(status: str, message: str, video_id: str | None = None, error: str | None = None) -> None:
+    def update_job(
+        status: str, message: str, video_id: str | None = None, error: str | None = None
+    ) -> None:
         try:
             job = db.query(IngestJob).filter(IngestJob.id == job_id).first()
             if job:
@@ -159,7 +161,9 @@ def update_job(status: str, message: str, video_id: str | None = None, error: st
                 db.commit()
         except Exception:
             db.rollback()
-            logger.warning(f"[Ingest] Job {job_id} — failed to update status to {status!r}")
+            logger.warning(
+                f"[Ingest] Job {job_id} — failed to update status to {status!r}"
+            )
 
     try:
         settings.ensure_dirs()
@@ -169,7 +173,11 @@ def update_job(status: str, message: str, video_id: str | None = None, error: st
         ingester = _get_ingester(source_type, platform, api_credentials)
         asset = ingester.ingest(source)
 
-        update_job("transcribing", "Transcribing audio with Whisper...", video_id=asset.video_id)
+        update_job(
+            "transcribing",
+            "Transcribing audio with Whisper...",
+            video_id=asset.video_id,
+        )
 
         # ── Step 2: Save video record ─────────────────────────────────────────
         # Use merge so a concurrent job for the same video_id doesn't raise IntegrityError.
@@ -191,11 +199,15 @@ def update_job(status: str, message: str, video_id: str | None = None, error: st
             db.rollback()
             video = db.query(Video).filter(Video.id == asset.video_id).first()
             if video is None:
-                raise RuntimeError(f"Video {asset.video_id} not found after merge") from None
+                raise RuntimeError(
+                    f"Video {asset.video_id} not found after merge"
+                ) from None
 
         # ── Step 3: Transcribe ────────────────────────────────────────────────
         transcriber = _get_transcriber()
-        transcript = transcriber.transcribe(asset.local_audio_path, asset.video_id, language)
+        transcript = transcriber.transcribe(
+            asset.local_audio_path, asset.video_id, language
+        )
 
         # Save transcript to disk
         transcript_path = Path(settings.transcript_dir) / f"{asset.video_id}.json"
@@ -210,7 +222,9 @@ def update_job(status: str, message: str, video_id: str | None = None, error: st
         chunks = chunker.chunk(transcript, asset)
 
         if not chunks:
-            raise ValueError("No chunks produced from transcript — audio may be silent or too short")
+            raise ValueError(
+                "No chunks produced from transcript — audio may be silent or too short"
+            )
 
         # ── Step 5: Embed ─────────────────────────────────────────────────────
         embedder = _get_embedder()
@@ -229,7 +243,11 @@ def update_job(status: str, message: str, video_id: str | None = None, error: st
             video.indexed_at = datetime.utcnow()
             db.commit()
 
-        update_job("done", f"Indexed {len(chunks)} chunks successfully", video_id=asset.video_id)
+        update_job(
+            "done",
+            f"Indexed {len(chunks)} chunks successfully",
+            video_id=asset.video_id,
+        )
         logger.success(f"[Ingest] Job {job_id} complete — {len(chunks)} chunks indexed")
 
     except Exception as e:
@@ -251,7 +269,9 @@ def update_job(status: str, message: str, video_id: str | None = None, error: st
 
 
 def _detect_source_type(source: str) -> str:
-    if source.startswith("http") and ("youtu" in source or "vimeo" in source or "twitch" in source):
+    if source.startswith("http") and (
+        "youtu" in source or "vimeo" in source or "twitch" in source
+    ):
         return "youtube"
     if source.startswith("rtmp") or ".m3u8" in source:
         return "live_stream"
@@ -260,7 +280,11 @@ def _detect_source_type(source: str) -> str:
     return "local_file"
 
 
-def _get_ingester(source_type: str, platform: str | None = None, credentials: dict[str, str] | None = None) -> Any:
+def _get_ingester(
+    source_type: str,
+    platform: str | None = None,
+    credentials: dict[str, str] | None = None,
+) -> Any:
     from ingestion import LiveStreamIngester, LocalFileIngester, YouTubeIngester
     from ingestion.video_api import get_api_ingester
 
@@ -271,6 +295,8 @@ def _get_ingester(source_type: str, platform: str | None = None, credentials: di
     elif source_type == "live_stream":
         return LiveStreamIngester(audio_dir=settings.audio_dir)
     elif source_type == "video_api":
-        return get_api_ingester(platform or "vimeo", credentials or {}, audio_dir=settings.audio_dir)
+        return get_api_ingester(
+            platform or "vimeo", credentials or {}, audio_dir=settings.audio_dir
+        )
     else:
         return YouTubeIngester(audio_dir=settings.audio_dir)

api/routes/query.py

@@ -46,8 +46,14 @@ def query_videos(request: QueryRequest) -> QueryResponse:
     )
 
 
-@router.post("/summarize", response_model=SummarizeResponse, responses={404: {"description": "Video not found"}})
-def summarize_video(request: SummarizeRequest, db: Annotated[Session, Depends(get_db)]) -> SummarizeResponse:
+@router.post(
+    "/summarize",
+    response_model=SummarizeResponse,
+    responses={404: {"description": "Video not found"}},
+)
+def summarize_video(
+    request: SummarizeRequest, db: Annotated[Session, Depends(get_db)]
+) -> SummarizeResponse:
     """
     Generate a summary of an indexed video.
     Uses map-reduce to handle long videos gracefully.
@@ -56,7 +62,9 @@ def summarize_video(request: SummarizeRequest, db: Annotated[Session, Depends(ge
 
     video = db.query(Video).filter(Video.id == request.video_id).first()
     if not video:
-        raise HTTPException(status_code=404, detail=f"Video {request.video_id!r} not found")
+        raise HTTPException(
+            status_code=404, detail=f"Video {request.video_id!r} not found"
+        )
 
     summarizer = Summarizer()
     result = summarizer.summarize(
@@ -68,7 +76,8 @@ def summarize_video(request: SummarizeRequest, db: Annotated[Session, Depends(ge
     chapter_summaries = None
     if result.chapter_summaries:
         chapter_summaries = [
-            ChapterSummary(chapter=c["chapter"], summary=c["summary"]) for c in result.chapter_summaries
+            ChapterSummary(chapter=c["chapter"], summary=c["summary"])
+            for c in result.chapter_summaries
         ]
 
     return SummarizeResponse(

config.py

@@ -89,7 +89,12 @@ class Config:
 
     def ensure_dirs(self) -> None:
         """Create all required data directories if they don't exist."""
-        for d in [self.data_dir, self.audio_dir, self.transcript_dir, self.chroma_persist_dir]:
+        for d in [
+            self.data_dir,
+            self.audio_dir,
+            self.transcript_dir,
+            self.chroma_persist_dir,
+        ]:
             Path(d).mkdir(parents=True, exist_ok=True)
 
 

ingestion/base.py

@@ -31,7 +31,9 @@ class VideoAsset:
     local_audio_path: Path  # Path to extracted 16kHz mono WAV
     duration_seconds: float | None = None
     description: str | None = None
-    chapters: list[dict[str, str | float]] = field(default_factory=list)  # [{title, start, end}]
+    chapters: list[dict[str, str | float]] = field(
+        default_factory=list
+    )  # [{title, start, end}]
     thumbnail_url: str | None = None
     uploader: str | None = None
     upload_date: str | None = None
@@ -50,7 +52,7 @@ class BaseIngester(ABC):
       3. Returns a VideoAsset ready for the transcription stage
     """
 
-    def __init__(self, audio_dir: str = "./data/audio"):
+    def __init__(self, audio_dir: str = "./data/audio") -> None:
         self.audio_dir = Path(audio_dir)
         self.audio_dir.mkdir(parents=True, exist_ok=True)
 

ingestion/live_stream.py

@@ -59,7 +59,10 @@ def ingest(self, source: str, video_id: str | None = None) -> VideoAsset:
             source_type=SourceType.LIVE_STREAM,
             local_audio_path=audio_path,
             duration_seconds=float(segment_seconds),
-            extra_metadata={"segment_type": "live", "capture_duration": segment_seconds},
+            extra_metadata={
+                "segment_type": "live",
+                "capture_duration": segment_seconds,
+            },
         )
 
     def ingest_stream(
@@ -93,7 +96,9 @@ def ingest_stream(
 
                 yield asset
 
-                logger.debug(f"[LiveStream] Segment {segment_count} done, capturing next...")
+                logger.debug(
+                    f"[LiveStream] Segment {segment_count} done, capturing next..."
+                )
 
         except KeyboardInterrupt:
             logger.info("[LiveStream] Capture stopped by user.")
@@ -103,7 +108,9 @@ def ingest_stream(
 
     # ── Private helpers ──────────────────────────────────────────────────────
 
-    def _capture_segment(self, source: str, output: Path, duration_seconds: int) -> None:
+    def _capture_segment(
+        self, source: str, output: Path, duration_seconds: int
+    ) -> None:
         """
         Use ffmpeg to read from the stream for `duration_seconds` and save as WAV.
         """

ingestion/local_file.py

@@ -77,7 +77,9 @@ def ingest(self, source: str, video_id: str | None = None) -> VideoAsset:
         self._extract_audio(source_path, audio_path)
 
         duration = float(metadata.get("format", {}).get("duration", 0)) or None
-        title = metadata.get("format", {}).get("tags", {}).get("title") or source_path.stem
+        title = (
+            metadata.get("format", {}).get("tags", {}).get("title") or source_path.stem
+        )
 
         logger.success(f"[LocalFile] Done → {audio_path}")
 
@@ -120,4 +122,6 @@ def _extract_audio(self, source: Path, output: Path) -> None:
                 .run(quiet=True)
             )
         except ffmpeg.Error as e:
-            raise RuntimeError(f"ffmpeg audio extraction failed: {e.stderr.decode()}") from e
+            raise RuntimeError(
+                f"ffmpeg audio extraction failed: {e.stderr.decode()}"
+            ) from e

ingestion/video_api.py

@@ -24,7 +24,9 @@ class VideoAPIIngester(BaseIngester):
     Subclasses implement _fetch_metadata() and _get_download_url().
     """
 
-    def __init__(self, credentials: dict[str, str], audio_dir: str = "./data/audio"):
+    def __init__(
+        self, credentials: dict[str, str], audio_dir: str = "./data/audio"
+    ) -> None:
         super().__init__(audio_dir)
         self.credentials = credentials
         self.session = requests.Session()
@@ -199,7 +201,9 @@ def _extract_twitch_vod_id(self, url: str) -> str:
 
     def _fetch_vod_metadata(self, vod_id: str) -> dict[str, Any]:
         try:
-            resp = self.session.get(f"{self.API_BASE}/videos", params={"id": vod_id}, timeout=30)
+            resp = self.session.get(
+                f"{self.API_BASE}/videos", params={"id": vod_id}, timeout=30
+            )
             resp.raise_for_status()
             data = resp.json().get("data", [])
             return data[0] if data else {}
@@ -247,7 +251,9 @@ def _download_via_ytdlp(self, url: str, output: Path) -> None:
 # ── Factory ──────────────────────────────────────────────────────────────────
 
 
-def get_api_ingester(platform: str, credentials: dict[str, str], audio_dir: str = "./data/audio") -> VideoAPIIngester:
+def get_api_ingester(
+    platform: str, credentials: dict[str, str], audio_dir: str = "./data/audio"
+) -> VideoAPIIngester:
     """
     Factory: return the right VideoAPIIngester for the given platform name.
 
@@ -265,5 +271,7 @@ def get_api_ingester(platform: str, credentials: dict[str, str], audio_dir: str
     }
     cls = registry.get(platform.lower())
     if cls is None:
-        raise ValueError(f"Unknown platform '{platform}'. Supported: {list(registry.keys())}")
+        raise ValueError(
+            f"Unknown platform '{platform}'. Supported: {list(registry.keys())}"
+        )
     return cls(credentials=credentials, audio_dir=audio_dir)