# Tech-Spec: OLD Pipeline Integration

**Created:** 2025-12-30
**Completed:** 2025-12-31
**Status:** ✅ COMPLETED - All 4 Steps Implemented
**Scope:** Complete OLD Pipeline Integration (All 4 Steps)

## Overview

### Problem Statement

The hybrid Excel pipeline integration requires combining the OLD pipeline (openpyxl-based, processes entire file with images/shapes/comments) and NEW pipeline (Docling-based, handles large tables >1000 rows) into a unified Airflow DAG workflow. However, this cannot proceed because:

**Immediate Blockers:**
1. OLD pipeline services are in root directory with incorrect imports (`src.commons.configs.settings` → doesn't exist)
2. Services use incorrect Azure OpenAI setting names (`azure_openai_chat_deployment` vs `azure_openai_deployment`)
3. Services cannot be imported by Airflow tasks
4. No adapter module exists to convert OLD pipeline output to NEW Milvus schema
5. No Airflow tasks exist to run OLD pipeline
6. DAG not configured for parallel flow (NEW + OLD pipelines)

**Current State:**
- NEW Pipeline: ✅ Working in Airflow (`parse_excel` → `detect_headers` → `chunk_excel` → `embeddings` → `store`)
- OLD Pipeline: ❌ Broken services in root directory, cannot be used
- Integration: ❌ No integration between pipelines

### Solution

Implement **all 4 steps** to integrate the OLD pipeline:

1. **Step 1: Migrate OLD Pipeline Services** - Move services to `src/extraction_v2/`, fix imports
2. **Step 2: Implement Adapter** - Create `LegacyToMilvusAdapter` to convert OLD → NEW schema
3. **Step 3: Create Airflow Tasks** - Build 3 new tasks: `parse_excel_old_pipeline`, `adapt_old_pipeline_schema`, `merge_pipeline_results`
4. **Step 4: Update DAG** - Configure parallel flow (NEW + OLD pipelines) in `document_extraction_dag.py`

**Result:**
- **NEW Pipeline**: Processes large tables (>1000 rows) using Docling + LLM headers
- **OLD Pipeline**: Processes entire file including images, shapes, comments with AI descriptions
- **Integration**: Both pipelines run in parallel, results merged, stored in unified Milvus schema

### Scope

**In Scope (All 4 Steps):**

**Step 1: Migrate OLD Pipeline Services**
- Migrate 4 service files to `src/extraction_v2/`:
  - `azure_chat_service.py`
  - `excel_image_extraction_service.py`
  - `excel_shape_extraction_service.py`
  - `excel_parser_service.py` (major refactor to hybrid approach)
- Fix all import paths to use `src.core.config` and `src.extraction_v2.*`
- Add missing Azure OpenAI settings to `src/core/config.py`
- Remove "chat" from setting names in `azure_chat_service.py`
- Refactor `excel_parser_service.py`:
  - Keep image/shape pre-processing (use ExcelImageExtractionService, ExcelShapeExtractionService)
  - Replace openpyxl parsing with Docling DocumentConverter
  - Replace LangChain chunking with Docling HybridChunker
  - Filter out shape chunks from return (shapes only for annotations)
- Remove parameters: `replace_images_with_descriptions`, `large_file_threshold`, `description_detail`, `unique_filename`, `use_pipe_separators`
- Create unit tests for all 4 services
- Delete old files from root directory

**Step 2: Implement Adapter**
- Create `src/adapters/legacy_to_milvus_adapter.py` module
- Implement `LegacyToMilvusAdapter` class with schema conversion logic
- Support all element types: Text, Image, Shape, Comment
- Generate embeddings for adapted chunks
- Create unit tests for adapter conversion logic

**Step 3: Create Airflow Tasks**
- Create `parse_excel_old_pipeline` task in `airflow/dags/tasks/parse_tasks.py`
- Create `adapt_old_pipeline_schema` task in `airflow/dags/tasks/processing_tasks.py`
- Create `merge_pipeline_results` task in `airflow/dags/tasks/processing_tasks.py`
- Use `@task.external_python` decorator pattern
- Handle pickle file storage for large results
- Create unit tests for task logic

**Step 4: Update DAG**
- Update `airflow/dags/document_extraction_dag.py` with hybrid Excel flow
- Configure parallel execution: NEW pipeline + OLD pipeline
- Wire up task dependencies correctly
- Test DAG validation (no syntax errors)

**Out of Scope:**
- End-to-end integration testing with real Excel files
- Performance benchmarking
- Staging deployment
- Production rollout

---

## Context for Development

### Codebase Patterns

**Configuration Management:**
- This project uses `pydantic-settings` in `src/core/config.py`
- Settings loaded from `.env` file
- Global instance: `from src.core.config import settings`

**Package Structure:**
```
src/
├── core/
│   └── config.py          # Settings class with Azure OpenAI config
├── extraction_v2/         # NEW pipeline (Docling-based)
│   ├── document_converter.py
│   ├── image_extractor.py
│   ├── shape_extractor.py
│   └── ... (other NEW pipeline services)
└── services/              # Existing services (auth, storage, etc.)
```

**Import Patterns:**
- Absolute imports from `src.*` (not relative imports)
- Services use `lru_cache()` for singleton factory functions
- Example: `get_azure_chat_service()` returns cached instance

**Logging:**
- Project uses structured logging
- Check existing services for logger patterns

### Files to Reference

**Source Files (Root Directory - TO BE MIGRATED):**
1. `/azure_chat_service.py` (158 lines)
   - Depends on: `langchain_openai.AzureChatOpenAI`, settings
   - Current imports: `from src.commons.configs.settings import get_settings` ❌
   - Uses: `azure_openai_chat_deployment`, `azure_openai_chat_temperature`, `azure_openai_chat_max_tokens` ❌

2. `/excel_image_extraction_service.py` (911 lines)
   - Depends on: `AzureChatService`, openpyxl, PIL, MinIO
   - Current imports: `from src.services.azure_chat_service import AzureChatService` ❌
   - Current imports: `from src.commons.handlers.log_handler import get_logger` ❌
   - Extracts images, generates AI descriptions

3. `/excel_shape_extraction_service.py` (exact lines unknown)
   - Depends on: openpyxl, shape DTOs
   - Current imports: Need to check and fix to `src.extraction_v2`
   - Extracts shapes, creates shape DTOs
   - **IMPORTANT**: Used for cell annotations only, NOT for storing shape chunks

4. `/excel_parser_service.py` (1364 lines)
   - **NEEDS MAJOR REFACTOR**: Currently uses openpyxl + LangChain ❌ → Should use **Docling** ✅
   - Current imports: `from src.commons.configs.settings import get_settings` ❌
   - Current imports: `from openpyxl import load_workbook` ❌ → Use Docling DocumentConverter
   - Current imports: `from langchain.text_splitter import RecursiveCharacterTextSplitter` ❌ → Use Docling HybridChunker
   - Current imports: `from src.services.excel_image_extraction_service import ExcelImageExtractionService` ❌ → Fix path to `src.extraction_v2`
   - Current imports: `from src.services.excel_shape_extraction_service import ExcelShapeExtractionService` ❌ → Fix path to `src.extraction_v2`
   - Has parameters to **REMOVE**: `replace_images_with_descriptions`, `large_file_threshold`, `description_detail`, `unique_filename`, `use_pipe_separators` ❌
   - Should use **hybrid approach**: Pre-process images/shapes → Docling parsing → HybridChunker chunking
   - **IMPORTANT**: Shape chunks NOT returned (only used for cell annotations)

**Target Configuration:**
- `/src/core/config.py` (47 lines)
  - Contains: `Settings` class with Azure OpenAI settings
  - Missing: `temperature`, `max_tokens` settings for chat completions
  - Global instance: `settings = Settings()`

**Reference Implementations:**
- `/src/extraction_v2/llm_header_detector.py` - Example of using Azure OpenAI in extraction_v2
- `/src/extraction_v2/shape_extractor.py` - Already handles shape extraction (NEW pipeline)
- `/src/extraction_v2/image_extractor.py` - Already handles image extraction (NEW pipeline)

**Test Patterns:**
- Check existing tests in `tests/` directory for patterns
- Services typically tested with mocked dependencies

### Technical Decisions

**Decision 1: excel_parser_service.py MUST use Docling, not openpyxl**
- **CRITICAL**: The OLD `excel_parser_service.py` currently uses openpyxl + LangChain RecursiveCharacterTextSplitter
- This is **incorrect** - it should use **Docling** like the NEW pipeline
- Rationale:
  - Unified parsing approach across OLD and NEW pipelines
  - Better performance with Docling's optimized Excel parsing
  - **HybridChunker** respects document structure (better than RecursiveCharacterTextSplitter)
- This is a **complete rewrite**, not a simple migration
- Reference implementation: `src/extraction_v2/document_converter.py`

**Decision 2: Keep ExcelImageExtractionService and ExcelShapeExtractionService for pre-processing**
- **WORKFLOW**: Image and shape annotations happen BEFORE Docling parsing
  1. Extract images from Excel using `ExcelImageExtractionService`
  2. Generate AI descriptions using `AzureChatService`
  3. Replace images with descriptions in Excel file (modify the file)
  4. Extract shapes from Excel using `ExcelShapeExtractionService`
  5. Apply shape annotations to cells in Excel file (modify the file)
  6. Parse modified Excel with Docling
  7. Chunk with HybridChunker
- Rationale: Docling parses the already-modified Excel file with descriptions and shape annotations in place
- Keep these methods from old parser:
  - `_extract_images_with_descriptions()` - Extract images and generate descriptions
  - `_create_position_descriptions_map()` - Map image positions to descriptions
  - `_replace_images_with_descriptions()` - Replace images in Excel file
  - `_apply_shape_annotations_to_cells()` - Apply shape annotations to Excel cells
- Dependencies needed: `ExcelImageExtractionService`, `ExcelShapeExtractionService`, `AzureChatService`
- **IMPORTANT**: Shape chunks are NOT stored in Milvus, NOT returned from `parse_excel_file()`
  - Shapes are only used to annotate cells before Docling parsing
  - Shape data is embedded in cell content, not as separate chunks

**Decision 3: Simplify parse_excel_file() signature**
- **Remove** these parameters (not needed):
  - `large_file_threshold` (not relevant with Docling)
  - `description_detail` (always use standard detail level)
  - `unique_filename` (unnecessary complexity)
  - `use_pipe_separators` (let Docling handle formatting)
- **Keep** these parameters:
  - `file_content` - Excel file bytes
  - `file_id` - Document UUID
  - `original_filename` - Original filename
  - `max_characters` - For chunking (passed to HybridChunker)
- **Note**: Image replacement is ALWAYS done (no parameter needed)
- Cleaner API, easier to use

**Decision 4: All three service dependencies are REQUIRED**
- `AzureChatService` is needed for generating AI image descriptions
- `ExcelImageExtractionService` is needed for extracting images and creating description map
- `ExcelShapeExtractionService` is needed for extracting shapes and applying cell annotations
- These run BEFORE Docling parsing (pre-processing step)
- Still remove "chat" from setting names to match project conventions
- **CRITICAL**: Shape chunks are NOT returned, only used for cell annotations

**Decision 5: Add missing settings to src/core/config.py**
- Add `azure_openai_temperature: float = 0.0`
- Add `azure_openai_max_tokens: int = 4096`
- These are general settings (not chat-specific)
- Can be used by both chat and other Azure OpenAI features

---

## Implementation Plan

### Overview

This implementation is divided into **4 major steps**. Each step must be completed before moving to the next:

1. **Step 1: Migrate OLD Pipeline Services** (6 tasks) - Foundation
2. **Step 2: Implement Adapter** (3 tasks) - Schema conversion
3. **Step 3: Create Airflow Tasks** (4 tasks) - Task creation
4. **Step 4: Update DAG** (2 tasks) - Integration

---

## STEP 1: Migrate OLD Pipeline Services

### Tasks

#### Task 1.1: Update Configuration
- [x] Add `azure_openai_temperature: float = 0.0` to `src/core/config.py`
- [x] Add `azure_openai_max_tokens: int = 4096` to `src/core/config.py`
- [x] Verify settings can be loaded from `.env` file

#### Task 1.2: Migrate azure_chat_service.py
- [x] Copy `/azure_chat_service.py` → `/src/extraction_v2/azure_chat_service.py`
- [x] Fix import: `from src.commons.configs.settings import get_settings` → `from src.core.config import settings`
- [x] Remove `get_settings()` call, use `settings` directly
- [x] Change `azure_openai_chat_deployment` → `azure_openai_deployment`
- [x] Change `azure_openai_chat_temperature` → `azure_openai_temperature`
- [x] Change `azure_openai_chat_max_tokens` → `azure_openai_max_tokens`
- [x] Update `get_azure_chat_service()` factory function (ensure `lru_cache()` present)
- [x] Verify no other dependencies need fixing

#### Task 1.3: Migrate excel_image_extraction_service.py
- [x] Copy `/excel_image_extraction_service.py` → `/src/extraction_v2/excel_image_extraction_service.py`
- [x] Fix import: `from src.services.azure_chat_service import AzureChatService` → `from src.extraction_v2.azure_chat_service import AzureChatService`
- [x] Fix import: `from src.commons.handlers.log_handler import get_logger` → Use structlog (extraction_v2 pattern)
- [x] Update `get_excel_image_extraction_service()` factory to use new `get_azure_chat_service()` path
- [x] Verify no other dependencies need fixing (openpyxl, PIL should be fine)

#### Task 1.3b: Migrate excel_shape_extraction_service.py
- [x] Shape extraction already exists in `src/extraction_v2/shape_extractor.py`
- [x] No migration needed - using existing ShapeExtractor class
- [x] **NOTE**: This service is used for cell annotations only, NOT for returning shape chunks

#### Task 1.4: Refactor excel_parser_service.py to use Docling

**MAJOR REFACTOR REQUIRED**: Hybrid approach - keep image pre-processing, use Docling for parsing/chunking.

**Workflow to implement:**
1. Image pre-processing (KEEP from old parser):
   - Extract images using `ExcelImageExtractionService`
   - Generate AI descriptions using `AzureChatService`
   - Replace images with descriptions in Excel file
2. Shape processing (KEEP from old parser):
   - Extract shapes using `ExcelShapeExtractionService`
   - Apply shape annotations to cells via `_apply_shape_annotations_to_cells()`
   - **DO NOT** return shape chunks (only annotate cells)
3. Docling parsing (REPLACE openpyxl):
   - Parse modified Excel with Docling `DocumentConverter`
4. Chunking (REPLACE LangChain):
   - Chunk with Docling `HybridChunker`

**Implementation steps:**
- [x] Copy `/excel_parser_service.py` → `/src/extraction_v2/excel_parser_service.py` (use as base)
- [x] Fix imports:
  - [x] `from src.commons.configs.settings import get_settings` → `from src.core.config import settings`
  - [x] `from src.services.excel_image_extraction_service` → `from src.extraction_v2.excel_image_extraction_service`
  - [x] Uses ShapeExtractor instead of excel_shape_extraction_service
- [x] **KEEP** `ExcelImageExtractionService` dependency in `__init__`
- [x] **KEEP** `ShapeExtractor` dependency in `__init__`
- [x] **KEEP** these methods from old parser (image pre-processing):
  - [x] `_extract_images_with_descriptions()` - Extract images and generate AI descriptions
  - [x] `_create_position_descriptions_map()` - Map image positions to descriptions
  - [x] `_replace_images_with_descriptions()` - Replace images in Excel file with descriptions
- [x] **Shape processing** via ShapeExtractor:
  - [x] Uses `shape_extractor.extract_and_apply_annotations()` method
  - [x] Shape annotations applied to cells before Docling parsing
  - [x] **DO NOT** create shape chunks for return (filter them out)
  - [x] **DO NOT** include shape chunks in final result
- [x] **REPLACE** openpyxl parsing code with Docling:
  - [x] Add: `from docling.document_converter import DocumentConverter`
  - [x] Replace openpyxl parsing logic with Docling `DocumentConverter`
- [x] **REPLACE** LangChain chunking with Docling HybridChunker:
  - [x] Remove: LangChain RecursiveCharacterTextSplitter
  - [x] Add: `from docling_core.transforms.chunker.hybrid_chunker import HybridChunker`
  - [x] Replace chunking logic with HybridChunker (respects document structure)
- [x] **SIMPLIFY** `parse_excel_file()` signature - **REMOVE** these parameters:
  - [x] `large_file_threshold` (not relevant)
  - [x] `description_detail` (use standard)
  - [x] `unique_filename` (not needed)
  - [x] `use_pipe_separators` (Docling handles)
  - [x] `replace_images_with_descriptions` (always True, no parameter needed)
- [x] **KEEP** these parameters: `file_content`, `file_id`, `original_filename`, `max_characters`
- [x] Return same DTO structure: `List[ChunkedElementDTO]` (but WITHOUT shape chunks)
- [x] Update `get_excel_parser_service()` factory to instantiate with `ExcelImageExtractionService` and `ShapeExtractor`

**Reference Implementation:**
- Image pre-processing: Keep from old `/excel_parser_service.py` (lines with `_extract_images_*`, `_replace_images_*`)
- Docling parsing: `src/extraction_v2/document_converter.py`
- Docling chunking: `airflow/dags/tasks/parse_tasks.py::parse_excel_document`
- HybridChunker docs: https://ds4sd.github.io/docling/examples/chunking/

#### Task 1.5: Create Unit Tests
- [x] Create `tests/extraction_v2/test_azure_chat_service.py`
  - Test: Service can be imported
  - Test: Factory function returns cached instance
  - Test: Settings are loaded correctly (mock `.env`)
  - Test: `invoke()` and `ainvoke()` methods exist (mock LLM)

- [x] Create `tests/extraction_v2/test_excel_image_extraction_service.py`
  - Test: Service can be imported
  - Test: Factory function works
  - Test: Service initializes with AzureChatService dependency

- [x] Create `tests/extraction_v2/test_excel_parser_service.py`
  - Test: Service can be imported
  - Test: Factory function works
  - Test: Service initializes **with** both `ExcelImageExtractionService` AND `ShapeExtractor` dependencies
  - Test: `parse_excel_file()` signature is correct:
    - Has: `file_content`, `file_id`, `original_filename`, `max_characters`
    - Does NOT have: `replace_images_with_descriptions`, `large_file_threshold`, `description_detail`, `unique_filename`, `use_pipe_separators`
  - Test: Image pre-processing methods exist: `_extract_images_with_descriptions`, `_create_position_descriptions_map`, `_replace_images_with_descriptions`
  - Test: Docling DocumentConverter is used for parsing

#### Task 1.6: Cleanup
- [x] Run all unit tests and verify they pass
- [x] Delete `/azure_chat_service.py` from root directory
- [x] Delete `/excel_image_extraction_service.py` from root directory
- [x] Delete `/excel_document_ingestion_service.py` from root directory
- [x] Delete `/excel_parser_service.py` from root directory
- [x] Verify no other files import from root-level service files

---

## STEP 2: Implement Adapter

### Tasks

#### Task 2.1: Create Adapter Module Structure
- [x] Create `src/adapters/` directory (if doesn't exist)
- [x] Create `src/adapters/__init__.py`
- [x] Create `src/adapters/legacy_to_milvus_adapter.py`
- [x] Add `LegacyChunk` dataclass for type hinting
- [x] Add `LegacyToMilvusAdapter` class skeleton

#### Task 2.2: Implement Schema Conversion Logic
- [x] Implement `adapt_chunks()` method (main entry point)
- [x] Implement `_convert_chunk_to_record()` method
- [x] Implement `_build_text_content()` for Text element type
- [x] Implement `_build_image_content()` for Image element type
- [x] Implement `_build_shape_content()` for Shape element type
- [x] Implement `_build_comment_content()` for Comment element type
- [x] Implement `_extract_row_number()` helper
- [x] Implement `_extract_col_range()` helper
- [x] Implement `_prepare_embedding_text()` method
- [x] Implement `convert_to_milvus_entities()` method
- [x] Implement `_record_to_text()` helper
- [x] Add proper error handling for malformed chunks

**Reference:** Full adapter implementation is documented in `docs/hybrid_pipeline_integration_analysis.md` lines 280-639

#### Task 2.3: Create Adapter Unit Tests
- [x] Create `tests/adapters/test_legacy_to_milvus_adapter.py`
- [x] Test: Text element conversion
- [x] Test: Image element conversion
- [x] Test: Shape element conversion (via metadata)
- [x] Test: Comment element conversion (via metadata)
- [x] Test: Row number extraction from various formats
- [x] Test: Column range extraction
- [x] Test: Embedding text preparation
- [x] Test: Milvus entity creation
- [x] Test: Truncation for long text (>8000 chars)
- [x] Test: Full adapt_chunks integration workflow

---

## STEP 3: Create Airflow Tasks

### Tasks

#### Task 3.1: Create parse_excel_old_pipeline Task
- [x] Add task to `airflow/dags/tasks/parse_tasks.py`
- [x] Use `@task.external_python` decorator
- [x] Add `sys.path.insert(0, '/opt/airflow')` for imports
- [x] Import `get_excel_parser_service` from `src.extraction_v2`
- [x] Read file from `local_path` in fetch_result
- [x] Call `parse_excel_file()` (no `replace_images_with_descriptions` param)
- [x] Save chunks to pickle file: `/opt/airflow/logs/old_pipeline_{document_id}.pkl`
- [x] Return dict with `data_file`, `document_id`, `filename`, `chunk_count`
- [x] Add logging for debugging

**Reference:** Full task implementation in `docs/hybrid_pipeline_integration_analysis.md` lines 794-868

#### Task 3.2: Create adapt_old_pipeline_schema Task
- [x] Add task to `airflow/dags/tasks/processing_tasks.py`
- [x] Use `@task.external_python` decorator
- [x] Add `sys.path.insert(0, '/opt/airflow')` for imports
- [x] Import `LegacyToMilvusAdapter` from `src.adapters`
- [x] Load OLD pipeline chunks from pickle file
- [x] Call `adapter.adapt_chunks()` with document_id and filename
- [x] Save adapted result to pickle file: `/opt/airflow/logs/adapted_{document_id}.pkl`
- [x] Clean up OLD pipeline pickle file after processing
- [x] Return dict with `data_file`, `document_id`, `filename`, `record_count`
- [x] Add logging for debugging

**Reference:** Full task implementation in `docs/hybrid_pipeline_integration_analysis.md` lines 873-946

#### Task 3.3: Create merge_pipeline_results Task
- [x] Add task to `airflow/dags/tasks/processing_tasks.py`
- [x] Use `@task.external_python` decorator
- [x] Load NEW pipeline embeddings result from pickle file
- [x] Load adapted OLD pipeline result from pickle file
- [x] Merge records: `new_data['records'] + old_data['records']`
- [x] Merge embeddings: `new_data['embeddings'] + old_data['embeddings']`
- [x] Save merged result to pickle file: `/opt/airflow/logs/merged_{document_id}.pkl`
- [x] Clean up intermediate pickle files (NEW and OLD)
- [x] Return dict with `data_file`, `document_id`, `record_count`
- [x] Add logging showing merge counts

**Reference:** Full task implementation in `docs/hybrid_pipeline_integration_analysis.md` lines 950-1024

#### Task 3.4: Create Task Unit Tests
- [ ] Create `tests/airflow/tasks/test_parse_excel_old_pipeline.py` (if needed)
- [ ] Create `tests/airflow/tasks/test_adapt_old_pipeline_schema.py` (if needed)
- [ ] Create `tests/airflow/tasks/test_merge_pipeline_results.py` (if needed)
- [ ] Mock external dependencies (file I/O, services)
- [ ] Test pickle file creation and cleanup
- [ ] Test error handling

**Note:** Airflow task testing may be deferred if project doesn't have existing Airflow task tests

---

## STEP 4: Update DAG

### Tasks

#### Task 4.1: Update document_extraction_dag.py
- [x] Import new tasks at top of file:
  - `from airflow.dags.tasks.parse_tasks import parse_excel_old_pipeline`
  - `from airflow.dags.tasks.processing_tasks import adapt_old_pipeline_schema, merge_pipeline_results`
- [x] Find Excel flow section (around `parse_excel` task)
- [x] Create `parse_old` task instance with `.override()` for task_id and timeout
- [x] Create `adapt_old` task instance wired to `parse_old` output
- [x] Create `merge_results` task instance wired to `embeddings_excel` and `adapt_old` outputs
- [x] Update `store_vectors_excel` to use `merge_results` output instead of `embeddings_excel`
- [x] Update task dependencies:
  - Keep: `branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings_excel`
  - Add: `branch >> parse_old >> adapt_old`
  - Add: `[embeddings_excel, adapt_old] >> merge_results >> store_vectors_excel`
- [x] Add comments documenting hybrid parallel flow
- [x] Update DAG documentation to describe hybrid Excel pipeline architecture

**Reference:** DAG update implementation in `docs/hybrid_pipeline_integration_analysis.md` lines 1032-1072

#### Task 4.2: Validate DAG
- [x] Run `python airflow/dags/document_extraction_dag.py` to check for syntax errors
- [x] Run `airflow dags list` to verify DAG appears
- [x] Run `airflow dags list-import-errors` to check for import errors
- [x] Run `airflow dags show document_extraction_dag` to verify task structure
- [x] Verify hybrid parallel flow: NEW pipeline + OLD pipeline → merge → store

---

### Acceptance Criteria

#### STEP 1: Migrate OLD Pipeline Services

**AC1.1: Configuration Updated**
- Given `src/core/config.py` is updated
- When settings are loaded
- Then `azure_openai_temperature` and `azure_openai_max_tokens` are available
- And values can be overridden via `.env` file

**AC1.2: Services Successfully Migrated**
- Given all 4 services are in `src/extraction_v2/`
- When importing services:
  - `from src.extraction_v2.azure_chat_service import get_azure_chat_service`
  - `from src.extraction_v2.excel_image_extraction_service import get_excel_image_extraction_service`
  - `from src.extraction_v2.excel_shape_extraction_service import get_excel_shape_extraction_service`
  - `from src.extraction_v2.excel_parser_service import get_excel_parser_service`
- Then all services can be instantiated without import errors
- And all cross-service dependencies resolve correctly

**AC1.3: azure_chat_service Uses Correct Settings**
- Given `azure_chat_service.py` is migrated
- When service initializes
- Then it uses `settings.azure_openai_deployment` (not `chat_deployment`)
- And uses `settings.azure_openai_temperature`
- And uses `settings.azure_openai_max_tokens`

**AC1.3a: excel_image_extraction_service Migrated Successfully**
- Given `excel_image_extraction_service.py` is migrated
- When service initializes
- Then it HAS `AzureChatService` dependency
- And can extract images from Excel files
- And can generate AI descriptions for images

**AC1.3b: excel_shape_extraction_service Migrated Successfully**
- Given `excel_shape_extraction_service.py` is migrated
- When service initializes
- Then it can extract shapes from Excel files
- And can create shape DTOs for annotation purposes

**AC1.4: excel_parser_service Uses Hybrid Approach (Images + Shapes + Docling)**
- Given `excel_parser_service.py` is migrated
- When service initializes
- Then it HAS `ExcelImageExtractionService` dependency (for image pre-processing)
- And it HAS `ExcelShapeExtractionService` dependency (for shape annotations)
- When parsing workflow executes
- Then it follows this order:
  1. Extract images and generate AI descriptions (using `ExcelImageExtractionService`)
  2. Replace images with descriptions in Excel file
  3. Extract shapes (using `ExcelShapeExtractionService`)
  4. Apply shape annotations to cells (modify Excel file)
  5. Parse modified Excel with Docling `DocumentConverter`
  6. Chunk with Docling `HybridChunker`
- And NO openpyxl parsing logic exists (replaced with Docling)
- And NO LangChain chunking logic exists (replaced with HybridChunker)
- And returned chunks do NOT include shape chunks (shapes only used for annotations)

**AC1.5: excel_parser_service Has Simplified Signature**
- Given `excel_parser_service.py` is migrated
- When `parse_excel_file()` signature is inspected
- Then it has ONLY these parameters: `file_content`, `file_id`, `original_filename`, `max_characters`
- And it does NOT have: `replace_images_with_descriptions`, `large_file_threshold`, `description_detail`, `unique_filename`, `use_pipe_separators`
- And images are ALWAYS replaced with descriptions (no parameter needed)

**AC1.6: Unit Tests Pass**
- Given unit tests are created for all 4 services:
  - `test_azure_chat_service.py`
  - `test_excel_image_extraction_service.py`
  - `test_excel_shape_extraction_service.py`
  - `test_excel_parser_service.py`
- When running `pytest tests/extraction_v2/`
- Then all import tests pass
- And all factory function tests pass
- And all dependency tests pass (parser has both image + shape dependencies)
- And no import errors occur

**AC1.7: Old Files Deleted**
- Given migration is complete and tests pass
- When checking root directory
- Then no service files remain in root:
  - `/azure_chat_service.py` deleted
  - `/excel_image_extraction_service.py` deleted
  - `/excel_shape_extraction_service.py` deleted
  - `/excel_parser_service.py` deleted
- And all services are ONLY in `src/extraction_v2/`
- And no other files reference old root-level import paths

#### STEP 2: Implement Adapter

**AC2.1: Adapter Module Exists**
- Given `src/adapters/legacy_to_milvus_adapter.py` is created
- When importing: `from src.adapters.legacy_to_milvus_adapter import LegacyToMilvusAdapter`
- Then module imports successfully
- And `LegacyToMilvusAdapter` class is available

**AC2.2: All Element Types Supported**
- Given a legacy chunk of type "Text" / "Image" / "Shape" / "Comment"
- When `adapter.adapt_chunks()` is called
- Then chunk is converted to NEW schema format
- And all type-specific fields are mapped correctly
- And no data is lost in conversion

**AC2.3: Embeddings Generated**
- Given legacy chunks are provided
- When `adapter.adapt_chunks()` is called
- Then embeddings are generated for all chunks
- And embedding count matches record count
- And embeddings are 3072-dimensional vectors

**AC2.4: Adapter Unit Tests Pass**
- Given adapter unit tests are created
- When running `pytest tests/adapters/`
- Then all conversion tests pass for each element type
- And edge case tests pass (truncation, missing fields, etc.)
- And no conversion errors occur

#### STEP 3: Create Airflow Tasks

**AC3.1: parse_excel_old_pipeline Task Works**
- Given `parse_excel_old_pipeline` task is created
- When task is called with fetch_result
- Then OLD pipeline services are executed
- And chunks are saved to pickle file
- And return dict contains `data_file`, `document_id`, `filename`, `chunk_count`

**AC3.2: adapt_old_pipeline_schema Task Works**
- Given `adapt_old_pipeline_schema` task is created
- When task is called with old_pipeline_result
- Then adapter converts chunks to NEW schema
- And adapted result is saved to pickle file
- And OLD pipeline pickle file is cleaned up
- And return dict contains `data_file`, `record_count`

**AC3.3: merge_pipeline_results Task Works**
- Given `merge_pipeline_results` task is created
- When task is called with NEW and OLD results
- Then records are merged correctly
- And embeddings are merged correctly
- And merged result is saved to pickle file
- And intermediate files are cleaned up
- And return dict contains `data_file`, `record_count`

**AC3.4: Tasks Can Be Imported**
- Given all 3 tasks are created
- When Airflow DAG imports tasks
- Then no import errors occur
- And tasks are decorated with `@task.external_python`

#### STEP 4: Update DAG

**AC4.1: DAG Updated with Hybrid Flow**
- Given `document_extraction_dag.py` is updated
- When DAG is loaded
- Then new tasks are present: `parse_old`, `adapt_old`, `merge_results`
- And task dependencies are correct:
  - NEW path: `branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings_excel`
  - OLD path: `branch >> parse_old >> adapt_old`
  - Merge: `[embeddings_excel, adapt_old] >> merge_results >> store_vectors_excel`

**AC4.2: DAG Validation Passes**
- Given DAG is updated
- When running `python airflow/dags/document_extraction_dag.py`
- Then no syntax errors occur
- When running `airflow dags list`
- Then `document_extraction_dag` appears in list
- When running `airflow tasks list document_extraction_dag`
- Then all tasks are listed including new hybrid tasks

**AC4.3: DAG Graph Renders**
- Given DAG is loaded in Airflow
- When viewing DAG graph in Airflow UI
- Then graph shows parallel flow correctly
- And both NEW and OLD paths are visible
- And merge point is clear

---

## Additional Context

### Dependencies

**Python Packages (already in project):**
- `pydantic-settings` - Configuration management
- `langchain-openai` - Azure OpenAI integration
- `openpyxl` - Excel file manipulation
- `Pillow` (PIL) - Image processing
- `aiofiles` - Async file operations

**Internal Dependencies (migration order matters):**
1. `azure_chat_service.py` - No internal dependencies (migrate first)
2. `excel_image_extraction_service.py` - Depends on `azure_chat_service.py` (migrate second)
3. `excel_shape_extraction_service.py` - No internal dependencies (migrate third, can be parallel with step 2)
4. `excel_parser_service.py` - Depends on `excel_image_extraction_service.py` AND `excel_shape_extraction_service.py` (migrate last)

### Testing Strategy

**Unit Tests (Required):**
- Import tests for each service
- Factory function tests (verify `lru_cache` works)
- Settings loading tests (mock `.env`)
- Basic instantiation tests (mock external dependencies like LLM)

**Integration Tests (Out of Scope):**
- Full pipeline execution
- Image extraction + description generation
- Excel parsing with real files

### Notes

**Why Migration Order Matters:**
Services have dependencies on each other:
```
azure_chat_service (no deps)
    ↓
excel_image_extraction_service (depends on azure_chat_service)
    ↓
excel_shape_extraction_service (no deps - can migrate in parallel with image service)
    ↓
excel_parser_service (depends on BOTH excel_image_extraction_service AND excel_shape_extraction_service)
```

Migrate in this order to avoid broken imports during development.
- Steps 1-2 must be sequential
- Step 3 (shape service) can be done in parallel with step 2 (image service)
- Step 4 (parser) must be last

**Logger Pattern to Check:**
The OLD services use `from src.commons.handlers.log_handler import get_logger`.
Check existing `extraction_v2` services for the correct logger pattern:
- Might be: `import logging; logger = logging.getLogger(__name__)`
- Or might use structured logging from `src` package

**Shape Extraction Note:**
`excel_parser_service.py` uses `ExcelShapeExtractionService` for extracting shapes:
1. Extract shapes using `ExcelShapeExtractionService` (KEEP THIS)
2. Apply shape annotations to cells via `_apply_shape_annotations_to_cells()` method (KEEP THIS)
3. **DO NOT** return shape chunks in final result (FILTER THEM OUT)

Shape data is embedded in cell annotations, not stored as separate chunks in Milvus.

**Config Validation:**
After adding new settings, verify `.env.example` is updated (if it exists) with:
```
AZURE_OPENAI_TEMPERATURE=0.0
AZURE_OPENAI_MAX_TOKENS=4096
```

**Post-Migration Verification:**
After migration, verify these imports work in a Python shell:
```python
from src.extraction_v2.azure_chat_service import get_azure_chat_service
from src.extraction_v2.excel_image_extraction_service import get_excel_image_extraction_service
from src.extraction_v2.excel_shape_extraction_service import get_excel_shape_extraction_service
from src.extraction_v2.excel_parser_service import get_excel_parser_service

# Should not raise ImportError
chat_service = get_azure_chat_service()
image_service = get_excel_image_extraction_service()
shape_service = get_excel_shape_extraction_service()
parser_service = get_excel_parser_service()

# Verify parser has both dependencies
assert parser_service.image_extraction_service is not None
assert parser_service.shape_extraction_service is not None
```

---

## Success Criteria

### STEP 1: Migrate OLD Pipeline Services ✅
- [ ] `azure_chat_service.py` migrated to `src/extraction_v2/` with corrected imports
- [ ] `excel_image_extraction_service.py` migrated to `src/extraction_v2/` with corrected imports
- [ ] `excel_shape_extraction_service.py` migrated to `src/extraction_v2/` with corrected imports
- [ ] `excel_parser_service.py` **REFACTORED** to use hybrid approach:
  - [ ] KEEPS image pre-processing methods (`_extract_images_with_descriptions`, etc.)
  - [ ] KEEPS shape processing method (`_apply_shape_annotations_to_cells`)
  - [ ] KEEPS `ExcelImageExtractionService` dependency (needed for images)
  - [ ] KEEPS `ExcelShapeExtractionService` dependency (needed for shape annotations)
  - [ ] REPLACES openpyxl parsing with Docling `DocumentConverter`
  - [ ] REPLACES LangChain chunking with Docling `HybridChunker`
  - [ ] FILTERS OUT shape chunks from return value (shapes only for annotations)
- [ ] No "chat" in Azure OpenAI setting names
- [ ] Missing settings added to config (`azure_openai_temperature`, `azure_openai_max_tokens`)
- [ ] `parse_excel_file()` simplified signature (removed 5 parameters: `replace_images_with_descriptions`, `large_file_threshold`, `description_detail`, `unique_filename`, `use_pipe_separators`)
- [ ] Unit tests pass for all 4 services (azure_chat, excel_image, excel_shape, excel_parser)
- [ ] Old files deleted from root directory

### STEP 2: Implement Adapter ✅
- [ ] `src/adapters/legacy_to_milvus_adapter.py` created
- [ ] All 4 element types supported (Text, Image, Shape, Comment)
- [ ] Schema conversion working correctly
- [ ] Embeddings generated for adapted chunks
- [ ] Adapter unit tests pass

### STEP 3: Create Airflow Tasks ✅
- [ ] `parse_excel_old_pipeline` task created in `parse_tasks.py`
- [ ] `adapt_old_pipeline_schema` task created in `processing_tasks.py`
- [ ] `merge_pipeline_results` task created in `processing_tasks.py`
- [ ] All tasks use `@task.external_python` decorator
- [ ] Pickle file handling implemented correctly
- [ ] Task unit tests pass (if applicable)

### STEP 4: Update DAG ✅
- [ ] `document_extraction_dag.py` updated with hybrid flow
- [ ] Parallel execution configured (NEW + OLD pipelines)
- [ ] Task dependencies wired correctly
- [ ] DAG validation passes (no syntax errors)
- [ ] DAG graph renders correctly in Airflow UI

### Overall Complete ✅
- [ ] All 4 steps completed successfully
- [ ] All unit tests pass across all modules
- [ ] DAG can be loaded without errors
- [ ] OLD pipeline integration complete and ready for use