# OVERVIEW

## 1) Architecture
The ingestion pipeline turns mixed file types into a single, traceable JSONL stream
for downstream chunking/embedding. It supports parallel processing via a
`ThreadPoolExecutor`-based engine and real-time progress streaming via Server-Sent
Events (SSE).

### Pipeline Flow
1. File discovery under an input root (`--input`), recursing into all subdirectories.
2. File classification: supported files are routed by extension; unsupported files
   (`.exe`, `.dll`, `.bin`, etc.) are flagged and reported.
3. Modality routing by extension:
   - Text: PDF/DOCX/PPTX/MD/HTML/TXT via LangChain + Unstructured loaders.
   - Images: OCR (Tesseract default, PaddleOCR optional).
   - Audio: faster-whisper transcription with timestamps.
   - Video: ffmpeg audio extraction, then faster-whisper.
   - CSV: row-wise text rendering.
4. Minimal normalization (whitespace + encoding cleanup only).
5. Emit JSONL records with stable `doc_id` and modality metadata.

### File Structure

```
01_data_ingestion/
├── ingest_multimodal.py          # CLI entry point + extraction functions
├── ingestion/
│   ├── __init__.py
│   ├── engine.py                 # IngestionEngine (parallel orchestrator)
│   ├── events.py                 # SSE event types + emitters
│   └── api.py                    # FastAPI router (upload, start, stream, cancel)
├── tests/
│   ├── conftest.py               # Shared fixtures
│   ├── test_engine.py            # Engine unit tests
│   ├── test_events.py            # Event system tests
│   ├── test_api.py               # API endpoint tests
│   └── test_integration.py       # End-to-end smoke tests
├── validate_ingestion.py         # Output schema validation
└── download_test_datasets.py     # Test data downloader
```

Output schema (one JSON object per line):
```json
{
  "doc_id": "policies-handbook-2024-pdf__a1b2c3d4",
  "page": 4,
  "section": "page_4",
  "section_index": 7,
  "content": "Employees are entitled to parental leave...",
  "source": {
    "path": "policies/handbook_2024.pdf",
    "filename": "handbook_2024.pdf",
    "filetype": "pdf",
    "modality": "text",
    "loader": "UnstructuredPDFLoader",
    "record_id": "policies-handbook-2024-pdf__a1b2c3d4:000007",
    "prev_record_id": "policies-handbook-2024-pdf__a1b2c3d4:000006",
    "next_record_id": "policies-handbook-2024-pdf__a1b2c3d4:000008"
  }
}
```

## 2) Frontend Integration Guide

This section gives frontend developers everything needed to integrate with the ingestion backend. The server runs on `http://localhost:8000` via `uvicorn server:app --reload --port 8000`.

### Starting a Job

**Option A — Server-side folder:**
```javascript
const res = await fetch("http://localhost:8000/ingestion/start", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ input_dir: "00_Data", workers: 4 }),
});
const { job_id, status } = await res.json();
// → { "job_id": "a1b2c3d4e5f6", "status": "running" }
```

**Option B — File upload (supports `.zip`):**
```javascript
const formData = new FormData();
formData.append("files", fileInput.files[0]);  // or a .zip
formData.append("workers", "4");

const res = await fetch("http://localhost:8000/ingestion/upload", {
  method: "POST",
  body: formData,
});
const { job_id, status, uploaded_files } = await res.json();
// → { "job_id": "a1b2c3d4e5f6", "status": "running", "uploaded_files": 15 }
```

### Streaming Progress (SSE)

Connect with `EventSource` to receive real-time events as files are processed:

```javascript
const evtSource = new EventSource(
  `http://localhost:8000/ingestion/${jobId}/stream`
);

evtSource.addEventListener("ingestion_started", (e) => {
  const data = JSON.parse(e.data);
  console.log(`Processing ${data.total_files} files`);
  // data.skipped_files[] lists unsupported formats that were flagged
});

evtSource.addEventListener("file_started", (e) => {
  const { file_path, file_index, total_files, modality } = JSON.parse(e.data);
  updateUI(`Processing ${file_path} (${file_index + 1}/${total_files})`);
});

evtSource.addEventListener("file_completed", (e) => {
  const { file_path, records_emitted, duration_sec } = JSON.parse(e.data);
  // duration_sec = wall-clock time for this individual file
});

evtSource.addEventListener("file_failed", (e) => {
  const { file_path, error, modality } = JSON.parse(e.data);
  showError(`${file_path}: ${error}`);
});

evtSource.addEventListener("file_skipped", (e) => {
  const { file_path, reason } = JSON.parse(e.data);
  // Unsupported file type flagged here
});

evtSource.addEventListener("ingestion_progress", (e) => {
  const { files_completed, total_files, records_emitted, elapsed_sec, eta_sec } =
    JSON.parse(e.data);
  updateProgressBar(files_completed / total_files);
  showETA(eta_sec);  // estimated seconds remaining
});

evtSource.addEventListener("ingestion_completed", (e) => {
  const { stats } = JSON.parse(e.data);
  // stats.counts.records_emitted, stats.duration_sec, stats.skipped_files[], etc.
  evtSource.close();
});
```

### Polling Fallback

If SSE drops or isn't supported, poll the status endpoint:

```javascript
async function pollStatus(jobId) {
  const res = await fetch(`http://localhost:8000/ingestion/${jobId}/status`);
  const data = await res.json();
  // → {
  //     "job_id": "...",
  //     "status": "running" | "completed" | "failed" | "cancelled",
  //     "files_completed": 12,
  //     "total_files": 47,
  //     "records_emitted": 892,
  //     "stats": null | { ... },    // populated when completed
  //     "error": null | "..."       // populated when failed
  //   }
  return data;
}
```

### Cancelling a Job

```javascript
await fetch(`http://localhost:8000/ingestion/${jobId}/cancel`, {
  method: "POST",
});
// Workers finish their current file, then stop cleanly.
// → { "job_id": "...", "status": "cancelling" }
```

### Response Schemas Reference

| Endpoint | Method | Response Fields |
|---|---|---|
| `/ingestion/start` | POST | `job_id`, `status` |
| `/ingestion/upload` | POST | `job_id`, `status`, `uploaded_files` |
| `/ingestion/{id}/stream` | GET | SSE stream (9 event types above) |
| `/ingestion/{id}/status` | GET | `job_id`, `status`, `files_completed`, `total_files`, `records_emitted`, `stats`, `error` |
| `/ingestion/{id}/cancel` | POST | `job_id`, `status` |

### Job Lifecycle

```
POST /start or /upload  →  pending  →  running  →  completed
                                          │
                                          ├──→  failed  (unrecoverable error)
                                          │
                             POST /cancel →  cancelling  →  cancelled
```

Jobs auto-expire from server memory 1 hour after completion.

---

## 3) Parallel Ingestion Engine

### Why Threads (not Processes)

The engine uses `ThreadPoolExecutor` rather than `ProcessPoolExecutor` because the
heavy workloads — Whisper inference (C/C++ via ctranslate2), FFmpeg (subprocess),
Tesseract (subprocess), and Unstructured PDF parsing (C libraries) — all release
Python's GIL during their compute phase. This means threads achieve true parallelism
for these operations without the overhead of inter-process serialization.

### How It Works

```
┌─────────────────────────────────────────────────────┐
│                  IngestionEngine.run()               │
│                                                     │
│  1. Discover files (rglob)                          │
│  2. Classify → supported[] + skipped[]              │
│  3. Submit to ThreadPoolExecutor(max_workers=N)     │
└────────────────────┬────────────────────────────────┘
                     │
        ┌────────────┼────────────┐
        ▼            ▼            ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │ Thread 1 │ │ Thread 2 │ │ Thread 3 │  ...N threads
   │          │ │          │ │          │
   │ PDF file │ │ CSV file │ │ image    │  ← run freely in parallel
   │ (no lock)│ │ (no lock)│ │ (no lock)│
   └─────────┘ └─────────┘ └─────────┘

   ┌─────────┐ ┌─────────┐
   │ Thread 4 │ │ Thread 5 │
   │          │ │          │
   │ audio    │ │ video    │  ← must acquire _whisper_lock
   │ (locked) │ │ (waits) │     only one Whisper job at a time
   └─────────┘ └─────────┘
```

### The Whisper Lock

Audio and video files require the Whisper model for transcription. Since:
- **CPU mode**: Whisper saturates all CPU cores — running two transcriptions in
  parallel would just thrash the CPU with no speed benefit.
- **GPU mode**: The model lives in VRAM — concurrent access would cause memory
  contention or OOM errors.

A `threading.Lock()` (`_whisper_lock`) serializes Whisper access. While one thread
is transcribing audio/video, other threads continue processing PDFs, images, CSVs,
and text files in parallel. This means a batch of 20 PDFs and 5 videos processes
all 20 PDFs concurrently while the 5 videos queue behind each other — the optimal
strategy for both CPU and GPU.

### Cooperative Cancellation

The engine uses a `threading.Event` (`_cancel`). When `engine.cancel()` is called:
1. The event is set.
2. Each thread checks `_cancel.is_set()` before starting a new file.
3. Threads finish their current file, then stop.
4. The executor shuts down with `cancel_futures=True` to drop queued work.

This gives clean shutdown — no half-written records, no orphaned temp files.

### Deterministic Output

Files are submitted to the thread pool with their original sorted index. As threads
complete in arbitrary order, results are collected as `(index, FileResult)` pairs.
Before writing to JSONL, results are sorted by index. This guarantees the output
is identical whether you use `--workers 0` (sequential) or `--workers 8` (parallel).

### Worker Count Guidelines

| `--workers` | Behavior |
|---|---|
| `0` | Sequential legacy mode (original `ingest()` function, for debugging) |
| `1` | Single thread through the new engine (useful for isolating issues) |
| `4` (default) | Good balance for most machines |
| `8+` | Benefits large batches of text/image files; Whisper is still serialized |

## 4) SSE Event System

### Event Types

| Event | Fields | When |
|---|---|---|
| `ingestion_started` | `job_id`, `total_files`, `skipped_files[]` | After file discovery |
| `file_started` | `file_path`, `file_index`, `total_files`, `modality`, `filetype` | File begins processing |
| `file_progress` | `file_path`, `detail`, `percent` | During long operations (video chunks) |
| `file_completed` | `file_path`, `records_emitted`, `duration_sec` | File done |
| `file_failed` | `file_path`, `error`, `modality` | File errored |
| `file_skipped` | `file_path`, `reason` | Unsupported format |
| `file_empty` | `file_path`, `modality`, `reason` | No content extracted |
| `ingestion_progress` | `files_completed`, `total_files`, `records_emitted`, `elapsed_sec`, `eta_sec` | After each file |
| `ingestion_completed` | `stats` | All done |

### Emitter Architecture

Events flow from worker threads to consumers via an emitter protocol:

```
Worker threads ──emit()──→ QueueEmitter ──stream()──→ SSE endpoint (async)
                           (thread-safe      (yields SSE-formatted
                            queue.Queue)      strings to StreamingResponse)
```

- **QueueEmitter**: Bridges sync threads to async HTTP. Workers call `emit()` which
  puts events on a `queue.Queue`. The async `stream()` generator polls the queue and
  yields SSE text.
- **LogEmitter**: Prints human-readable progress to stdout (for CLI mode).
- **NullEmitter**: No-op (zero overhead for programmatic use without progress).
- **MultiEmitter**: Broadcasts to multiple emitters simultaneously.

### SSE Wire Format

```
event: file_completed
data: {"event_type": "file_completed", "file_path": "docs/report.pdf", "records_emitted": 45, "duration_sec": 2.3}

event: ingestion_progress
data: {"event_type": "ingestion_progress", "files_completed": 12, "total_files": 47, "records_emitted": 892, "elapsed_sec": 34.5, "eta_sec": 98.2}
```

## 5) HTTP API

All endpoints are mounted under `/ingestion/` on the FastAPI server.

### `POST /ingestion/start`
Start ingestion from a server-side directory.
```json
// Request
{"input_dir": "/path/to/folder", "workers": 4, "pdf_strategy": "auto"}
// Response
{"job_id": "a1b2c3d4e5f6", "status": "running"}
```

### `POST /ingestion/upload`
Upload files (or a `.zip`) via multipart form data.
```bash
curl -F "files=@documents.zip" -F "workers=4" http://localhost:8000/ingestion/upload
```
```json
// Response
{"job_id": "a1b2c3d4e5f6", "status": "running", "uploaded_files": 15}
```

### `GET /ingestion/{job_id}/stream`
SSE endpoint for real-time progress. Connect with `EventSource` in JavaScript or
`curl -N`.

### `GET /ingestion/{job_id}/status`
Polling fallback if the SSE connection drops.
```json
{"job_id": "...", "status": "running", "files_completed": 12, "total_files": 47, "records_emitted": 892}
```

### `POST /ingestion/{job_id}/cancel`
Cancel a running job. Workers stop after their current file.

## 6) Decisions and Justifications
- LangChain + Unstructured: widest coverage for document formats and metadata.
- Tesseract OCR default: light, local, reliable baseline.
- PaddleOCR optional: higher OCR fidelity on complex layouts; heavier install.
- faster-whisper for ASR: strong local quality, good speed, timestamped segments.
- Stable `doc_id`: slug of relative path + short hash; deterministic and traceable.
- Minimal cleanup only: avoid semantic changes before chunking/embedding.
- ThreadPoolExecutor over ProcessPoolExecutor: GIL is released by C extensions and
  subprocesses; threads avoid serialization overhead and share the Whisper model.
- Whisper lock: serializes transcription to avoid CPU thrashing or GPU OOM.
- Cooperative cancellation: clean shutdown without half-written output.

## 7) Reproduce Tests (Mac + Windows)
These steps reproduce the exact smoke tests used in this repo. They run each modality
separately to keep runtime small.

### Run the test suite
```bash
cd 01_data_ingestion
python -m pytest tests/ -v
```

### macOS
System deps:
```bash
brew install tesseract poppler ffmpeg
```

Python 3.11 venv (required for faster-whisper on macOS):
```bash
brew install python@3.11
python3.11 -m venv Final_venv
source Final_venv/bin/activate
pip install -r 01_data_ingestion/requirements.txt
pip install datasets
```

Download datasets (lightweight video subset is default):
```bash
python 01_data_ingestion/download_test_datasets.py \
  --out 00_Data_test \
  --text --image --audio --video \
  --cord-max-items 50
```

Run ingestion (smoke test):
```bash
# Text
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/text \
  --output-dir 01_data_ingestion/output/test_text

# Images (OCR sample of 5)
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/image/cord-v2/images \
  --output-dir 01_data_ingestion/output/test_image \
  --max-files 5

# Audio (small subset)
mkdir -p 00_Data_test/audio_subset
find 00_Data_test/audio/librispeech -name "*.flac" | head -n 2 | xargs -I{} cp "{}" 00_Data_test/audio_subset/
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/audio_subset \
  --output-dir 01_data_ingestion/output/test_audio

# Video (2 files)
python 01_data_ingestion/ingest_multimodal.py \
  --input 00_Data_test/video/yt-ugc-subset \
  --output-dir 01_data_ingestion/output/test_video \
  --max-files 2
```

Validate outputs:
```bash
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_text/cleaned_documents.jsonl
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_image/cleaned_documents.jsonl
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_audio/cleaned_documents.jsonl
python 01_data_ingestion/validate_ingestion.py --input 01_data_ingestion/output/test_video/cleaned_documents.jsonl
```

### Windows (PowerShell)
System deps (Chocolatey):
```powershell
choco install tesseract poppler ffmpeg
```

Python 3.11 venv:
```powershell
py -3.11 -m venv Final_venv
.\Final_venv\Scripts\Activate.ps1
pip install -r 01_data_ingestion/requirements.txt
pip install datasets
```

Download datasets:
```powershell
python 01_data_ingestion\download_test_datasets.py `
  --out 00_Data_test `
  --text --image --audio --video `
  --cord-max-items 50
```

Run ingestion (smoke test):
```powershell
# Text
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\text `
  --output-dir 01_data_ingestion\output\test_text

# Images (OCR sample of 5)
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\image\cord-v2\images `
  --output-dir 01_data_ingestion\output\test_image `
  --max-files 5

# Audio (small subset)
New-Item -ItemType Directory -Force -Path 00_Data_test\audio_subset | Out-Null
Get-ChildItem -Recurse -Path 00_Data_test\audio\librispeech -Filter *.flac | Select-Object -First 2 | Copy-Item -Destination 00_Data_test\audio_subset
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\audio_subset `
  --output-dir 01_data_ingestion\output\test_audio

# Video (2 files)
python 01_data_ingestion\ingest_multimodal.py `
  --input 00_Data_test\video\yt-ugc-subset `
  --output-dir 01_data_ingestion\output\test_video `
  --max-files 2
```

Validate outputs:
```powershell
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_text\cleaned_documents.jsonl
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_image\cleaned_documents.jsonl
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_audio\cleaned_documents.jsonl
python 01_data_ingestion\validate_ingestion.py --input 01_data_ingestion\output\test_video\cleaned_documents.jsonl
```

## Notes
- Default video URLs are direct `.webm` files; no `yt-dlp` required.
- Audio/video ingestion requires Python 3.10-3.12 on macOS due to `onnxruntime`.
- OCR can return empty text; those files are skipped to avoid noisy records.
- The `--workers 0` flag falls back to the original sequential `ingest()` function
  for backward compatibility and debugging.