"""
Parse document tasks using @task.external_python decorator.

These tasks run in the uv virtual environment at /opt/airflow/.venv
which has all the required dependencies (openpyxl, docling, etc.)

Large parse results are saved to pickle files to avoid XCom size limits.
Only small metadata (file paths, counts) are returned via XCom.
"""

from datetime import timedelta

from airflow.decorators import task
from typing import Dict, Any

# Import configuration - note: external_python tasks can't import from config
# due to different Python environment, so we keep these constants local
EXTERNAL_PYTHON = '/opt/airflow/.venv/bin/python'
PARSE_DATA_DIR = '/opt/airflow/logs'

# Retry policy: parsers are the only tasks allowed to retry (network/IO flaky).
# All other DAG tasks inherit retries=0 from dag_config.default_args.
_PARSER_RETRIES = dict(
    retries=2,
    retry_delay=timedelta(seconds=30),
    retry_exponential_backoff=True,
)


@task.external_python(python=EXTERNAL_PYTHON, **_PARSER_RETRIES)
def parse_excel_large_tables(
    fetch_result: Dict[str, Any],
    tenant_id: str,
    node_id: str,
    correlation_id: str,
) -> Dict[str, Any]:
    """
    Parse Excel document for LARGE TABLES using Docling (NEW pipeline).

    This pipeline only processes large tables (>1000 rows) efficiently.
    All imports are inside the function to run in the external environment.
    Large results are saved to pickle file to avoid XCom size limits.

    Args:
        fetch_result: Result from fetch_document task containing local_path and document_id
        tenant_id: Tenant UUID for pipeline-event routing
        node_id: Node UUID for pipeline-event routing
        correlation_id: Airflow dag_run_id, for event correlation across stages

    Returns:
        Dict with data_file path and metadata (actual data stored in pickle file)
    """
    # All imports must be inside the function for external_python
    from src.extraction_v2.document_converter import DocumentToHtmlConverter
    from src.extraction_v2.html_table_extractor import HtmlTableExtractor
    from src.extraction_v2.large_table_extractor import LargeTableExtractor
    from src.extraction_v2.excel_conversion_utils import convert_xls_to_xlsx
    from src.extraction_v2.strikethrough_utils import remove_strikethrough_cells
    import logging
    import pickle
    import os
    import shutil
    import tempfile

    log = logging.getLogger(__name__)

    if not fetch_result:
        raise ValueError("No fetch result provided")

    local_path = fetch_result['local_path']
    document_id = fetch_result['document_id']
    original_filename = fetch_result.get('original_filename', '')

    from src.services.pipeline_events import publish_stage_event
    publish_stage_event(
        "parsing", "started",
        document_id=document_id, tenant_id=tenant_id,
        node_id=node_id, correlation_id=correlation_id,
    )

    log.info(f"Parsing Excel document {document_id}, path: {local_path}")

    # Check for password-protected files before processing
    from src.extraction_v2.password_detection import check_password_protected
    check_password_protected(local_path)

    # Copy file to task-specific temp directory to avoid race condition
    # with parse_excel_detailed running in parallel
    task_temp_dir = tempfile.mkdtemp(prefix=f'parse_large_{document_id}_')
    task_local_path = os.path.join(task_temp_dir, os.path.basename(local_path))
    shutil.copy2(local_path, task_local_path)
    log.info(f"Copied file to task-specific path: {task_local_path}")
    local_path = task_local_path

    # Step 1: Convert .xls to .xlsx if needed (preprocessing)
    xlsx_path, was_converted = convert_xls_to_xlsx(local_path)
    if was_converted:
        log.info(f"Converted .xls to .xlsx: {xlsx_path}")
        local_path = xlsx_path

    # Step 2: Remove strikethrough text (preprocessing)
    strikethrough_result = remove_strikethrough_cells(local_path)
    if strikethrough_result['cells_modified'] > 0:
        log.info(
            f"Removed strikethrough text: {strikethrough_result['runs_removed']} runs "
            f"from {strikethrough_result['cells_modified']} cells across "
            f"{strikethrough_result['sheets_affected']} sheets"
        )

    # Use the cleaned file for parsing
    local_path = strikethrough_result['output_file']

    converter = DocumentToHtmlConverter()

    # Excel: Large table detection matching pipeline.py flow
    log.info("Step 1: Detecting tables via border detection")
    large_extractor = LargeTableExtractor()
    detected_tables = large_extractor.detect_tables(local_path)

    large_tables = [t for t in detected_tables if t.get('is_large', False)]
    normal_tables = [t for t in detected_tables if not t.get('is_large', False)]

    log.info(
        f"Table classification: {len(large_tables)} large (>1000 rows), "
        f"{len(normal_tables)} normal"
    )

    large_table_records = []

    # Step 2: Process large tables with chunked extraction
    if large_tables:
        log.info(
            f"Step 2: Processing {len(large_tables)} large table(s) "
            "with chunked extraction"
        )
        for table in large_tables:
            result = large_extractor.extract_table(local_path, table)
            if result.success:
                # Attach sheet metadata to each record so we know which sheet it came from
                for idx, record in enumerate(result.records):
                    if isinstance(record, dict):
                        record['_sheet_name'] = result.sheet_name
                        record['_row_number'] = idx + 1  # 1-indexed row number within the table

                large_table_records.extend(result.records)
                log.info(
                    f"Large table extracted: {result.rows_processed} rows "
                    f"from sheet '{result.sheet_name}'"
                )
            else:
                log.warning(
                    f"Large table extraction failed for sheet "
                    f"'{table.get('sheet', 'unknown')}': {result.error}"
                )

    # Step 3: SKIP normal tables processing (NEW pipeline only handles large tables)
    # Normal tables, raw text tables, images, shapes, and comments are handled by OLD pipeline
    log.info("Step 3: Skipping normal tables (NEW pipeline handles large tables only)")

    log.info(
        f"Excel parsing complete: {len(large_table_records)} large table records"
    )

    # Save large data to pickle file to avoid XCom size limits
    parse_data = {
        'doc_type': 'excel',
        'large_table_records': large_table_records,
        'local_path': local_path,
        'document_id': document_id,
        'original_filename': original_filename,
    }

    data_file = f"/opt/airflow/logs/parse_excel_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(parse_data, f)

    log.info(f"Saved parse result to {data_file}")

    # Return only small metadata via XCom
    return {
        'data_file': data_file,
        'doc_type': 'excel',
        'document_id': document_id,
        'original_filename': original_filename,
        'local_path': local_path,
        'large_table_record_count': len(large_table_records),
    }


@task.external_python(python=EXTERNAL_PYTHON, **_PARSER_RETRIES)
def parse_word_document(
    fetch_result: Dict[str, Any],
    tenant_id: str,
    node_id: str,
    correlation_id: str,
) -> Dict[str, Any]:
    """
    Parse Word document using external Python environment.

    For .doc/.docm files, converts to .docx once and saves the converted file
    for reuse by chunk_text_task to avoid double conversion.
    Large results are saved to pickle file to avoid XCom size limits.

    Args:
        fetch_result: Result from fetch_document task
        tenant_id: Tenant UUID for pipeline-event routing
        node_id: Node UUID for pipeline-event routing
        correlation_id: Airflow dag_run_id, for event correlation across stages

    Returns:
        Dict with data_file path and metadata (actual data stored in pickle file)
    """
    from src.extraction_v2.document_converter import DocumentToHtmlConverter
    from pathlib import Path
    import logging
    import os
    import shutil
    import pickle

    log = logging.getLogger(__name__)

    if not fetch_result:
        raise ValueError("No fetch result provided")

    local_path = fetch_result['local_path']
    document_id = fetch_result['document_id']
    original_filename = fetch_result.get('original_filename', 'document.docx')

    from src.services.pipeline_events import publish_stage_event
    publish_stage_event(
        "parsing", "started",
        document_id=document_id, tenant_id=tenant_id,
        node_id=node_id, correlation_id=correlation_id,
    )

    log.info(f"Parsing Word document {document_id}, path: {local_path}")

    # Check for password-protected files before processing
    from src.extraction_v2.password_detection import check_password_protected
    check_password_protected(local_path)

    # Check if we need to convert .doc/.docm
    file_path = Path(local_path)
    converted_path = None

    if file_path.suffix.lower() in ['.doc', '.docm']:
        log.info(f"Converting {file_path.suffix} to .docx for reuse...")
        log.info(f"Original file path: {local_path}")
        log.info(f"Original file exists: {os.path.exists(local_path)}")

        converter_instance = DocumentToHtmlConverter()
        temp_docx = converter_instance._convert_doc_to_docx(file_path)

        if temp_docx is None:
            raise ValueError(f"Failed to convert {file_path.suffix} to .docx")

        log.info(f"Temporary converted file: {temp_docx}")
        log.info(f"Temporary converted file exists: {os.path.exists(temp_docx)}")

        # Save converted file to a persistent location for chunk_text_task
        persistent_dir = os.path.dirname(local_path)
        converted_path = os.path.join(persistent_dir, f"{file_path.stem}_converted.docx")
        shutil.copy2(temp_docx, converted_path)
        log.info(f"Saved converted .docx for reuse: {converted_path}")
        log.info(f"Saved converted file exists: {os.path.exists(converted_path)}")
        log.info(f"Saved converted file size: {os.path.getsize(converted_path)} bytes")

        # Clean up the temp conversion directory
        temp_dir = os.path.dirname(temp_docx)
        if temp_dir and os.path.exists(temp_dir) and 'tmp' in temp_dir:
            log.info(f"Cleaning up temp conversion directory: {temp_dir}")
            shutil.rmtree(temp_dir)

    # Convert using Docling with custom serializers
    from docling.document_converter import DocumentConverter
    from docling_core.transforms.serializer.markdown import (
        MarkdownDocSerializer,
        MarkdownParams,
        MarkdownTableSerializer,
    )
    from src.extraction_v2.semantic_chunker import AnnotationPictureSerializer

    file_to_convert = converted_path if converted_path else local_path
    docling_converter = DocumentConverter()
    result = docling_converter.convert(file_to_convert)
    docling_document = result.document

    # Export per-page markdown using custom serializers
    num_pages = docling_document.num_pages() if hasattr(docling_document, 'num_pages') and callable(docling_document.num_pages) else 0
    pages = []

    if num_pages > 0:
        for page_no in range(1, num_pages + 1):
            try:
                serializer = MarkdownDocSerializer(
                    doc=docling_document,
                    picture_serializer=AnnotationPictureSerializer(),
                    table_serializer=MarkdownTableSerializer(),
                    params=MarkdownParams(pages={page_no}),
                )
                page_md = serializer.serialize().text
                if page_md and page_md.strip():
                    pages.append({"page_number": page_no, "markdown": page_md})
            except Exception as e:
                log.warning(f"Failed to serialize page {page_no}: {e}")
    else:
        # Fallback: export full document
        serializer = MarkdownDocSerializer(
            doc=docling_document,
            picture_serializer=AnnotationPictureSerializer(),
            table_serializer=MarkdownTableSerializer(),
        )
        markdown = serializer.serialize().text
        if markdown:
            pages.append({"page_number": 1, "markdown": markdown})

    if not pages:
        raise ValueError(f"Failed to convert Word document: {local_path}")

    total_chars = sum(len(p["markdown"]) for p in pages)
    log.info(f"Successfully converted Word document: {len(pages)} pages, {total_chars} characters")

    # Save large data to pickle file to avoid XCom size limits
    parse_data = {
        'doc_type': 'word',
        'pages': pages,
        'local_path': local_path,
        'converted_path': converted_path or '',
        'document_id': document_id,
        'original_filename': original_filename,
    }

    data_file = f"/opt/airflow/logs/parse_word_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(parse_data, f)

    log.info(f"Saved parse result to {data_file}")

    # Return only small metadata via XCom
    return {
        'data_file': data_file,
        'doc_type': 'word',
        'document_id': document_id,
        'original_filename': original_filename,
        'local_path': local_path,
        'converted_path': converted_path or '',
        'page_count': len(pages),
        'content_length': total_chars,
    }


@task.external_python(python=EXTERNAL_PYTHON, **_PARSER_RETRIES)
def parse_pdf_document(
    fetch_result: Dict[str, Any],
    tenant_id: str,
    node_id: str,
    correlation_id: str,
) -> Dict[str, Any]:
    """
    Parse PDF document using external Python environment.
    Large results are saved to pickle file to avoid XCom size limits.

    Args:
        fetch_result: Result from fetch_document task
        tenant_id: Tenant UUID for pipeline-event routing
        node_id: Node UUID for pipeline-event routing
        correlation_id: Airflow dag_run_id, for event correlation across stages

    Returns:
        Dict with data_file path and metadata (actual data stored in pickle file)
    """
    from src.extraction_v2.document_converter import DocumentToHtmlConverter
    import logging
    import pickle

    log = logging.getLogger(__name__)

    if not fetch_result:
        raise ValueError("No fetch result provided")

    local_path = fetch_result['local_path']
    document_id = fetch_result['document_id']
    original_filename = fetch_result.get('original_filename', '')

    from src.services.pipeline_events import publish_stage_event
    publish_stage_event(
        "parsing", "started",
        document_id=document_id, tenant_id=tenant_id,
        node_id=node_id, correlation_id=correlation_id,
    )

    log.info(f"Parsing PDF document {document_id}, path: {local_path}")

    # Check for password-protected files before processing
    from src.extraction_v2.password_detection import check_password_protected
    check_password_protected(local_path)

    # Convert using Docling
    from docling.datamodel.base_models import InputFormat
    from docling.datamodel.pipeline_options import PdfPipelineOptions
    from docling.document_converter import DocumentConverter, PdfFormatOption
    from docling.exceptions import ConversionError
    from docling.backend.pypdfium2_backend import PyPdfiumDocumentBackend
    from docling_core.transforms.serializer.markdown import (
        MarkdownDocSerializer,
        MarkdownParams,
        MarkdownTableSerializer,
    )
    from docling_core.types.doc.labels import DocItemLabel
    from src.extraction_v2.semantic_chunker import AnnotationPictureSerializer

    pdf_pipeline_options = PdfPipelineOptions(do_ocr=False)
    converter = DocumentConverter(
        format_options={
            InputFormat.PDF: PdfFormatOption(pipeline_options=pdf_pipeline_options),
        }
    )

    try:
        result = converter.convert(local_path)
    except ConversionError:
        # docling-parse cannot handle PDFs with non-UTF-8 text encoding
        # (e.g. Shift-JIS in Japanese PDFs). Fall back to pypdfium2 backend
        # which handles these encodings correctly, at the cost of less
        # precise table column detection.
        log.warning(f"Default PDF backend failed, retrying with pypdfium2: {local_path}")
        fallback_converter = DocumentConverter(
            format_options={
                InputFormat.PDF: PdfFormatOption(
                    pipeline_options=pdf_pipeline_options,
                    backend=PyPdfiumDocumentBackend,
                ),
            }
        )
        result = fallback_converter.convert(local_path)

    docling_document = result.document

    # Export per-page markdown using custom serializers
    num_pages = docling_document.num_pages() if hasattr(docling_document, 'num_pages') and callable(docling_document.num_pages) else 0
    pages = []

    if num_pages > 0:
        for page_no in range(1, num_pages + 1):
            try:
                serializer = MarkdownDocSerializer(
                    doc=docling_document,
                    picture_serializer=AnnotationPictureSerializer(),
                    table_serializer=MarkdownTableSerializer(),
                    params=MarkdownParams(pages={page_no}),
                )
                page_md = serializer.serialize().text
                if page_md and page_md.strip():
                    pages.append({"page_number": page_no, "markdown": page_md})
            except Exception as e:
                log.warning(f"Failed to serialize page {page_no}: {e}")
    else:
        # Fallback: export full document
        serializer = MarkdownDocSerializer(
            doc=docling_document,
            picture_serializer=AnnotationPictureSerializer(),
            table_serializer=MarkdownTableSerializer(),
        )
        markdown = serializer.serialize().text
        if markdown:
            pages.append({"page_number": 1, "markdown": markdown})

    if not pages:
        raise ValueError(f"Failed to convert PDF document: {local_path}")

    total_chars = sum(len(p["markdown"]) for p in pages)
    log.info(f"Successfully converted PDF: {len(pages)} pages, {total_chars} characters")

    # Save large data to pickle file to avoid XCom size limits
    parse_data = {
        'doc_type': 'pdf',
        'pages': pages,
        'local_path': local_path,
        'document_id': document_id,
        'original_filename': original_filename,
    }

    data_file = f"/opt/airflow/logs/parse_pdf_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(parse_data, f)

    log.info(f"Saved parse result to {data_file}")

    # Return only small metadata via XCom
    return {
        'data_file': data_file,
        'doc_type': 'pdf',
        'document_id': document_id,
        'original_filename': original_filename,
        'local_path': local_path,
        'page_count': len(pages),
        'content_length': total_chars,
    }


@task.external_python(python=EXTERNAL_PYTHON, **_PARSER_RETRIES)
def parse_powerpoint_document(
    fetch_result: Dict[str, Any],
    tenant_id: str,
    node_id: str,
    correlation_id: str,
) -> Dict[str, Any]:
    """
    Parse PowerPoint document using external Python environment.
    Large results are saved to pickle file to avoid XCom size limits.

    Args:
        fetch_result: Result from fetch_document task
        tenant_id: Tenant UUID for pipeline-event routing
        node_id: Node UUID for pipeline-event routing
        correlation_id: Airflow dag_run_id, for event correlation across stages

    Returns:
        Dict with data_file path and metadata (actual data stored in pickle file)
    """
    from src.extraction_v2.document_converter import DocumentToHtmlConverter
    import logging
    import pickle

    log = logging.getLogger(__name__)

    if not fetch_result:
        raise ValueError("No fetch result provided")

    local_path = fetch_result['local_path']
    document_id = fetch_result['document_id']
    original_filename = fetch_result.get('original_filename', 'document.pptx')

    from src.services.pipeline_events import publish_stage_event
    publish_stage_event(
        "parsing", "started",
        document_id=document_id, tenant_id=tenant_id,
        node_id=node_id, correlation_id=correlation_id,
    )

    log.info(f"Parsing PowerPoint document {document_id}, path: {local_path}")

    # Check for password-protected files before processing
    from src.extraction_v2.password_detection import check_password_protected
    check_password_protected(local_path)

    # Convert using Docling with custom serializers
    from docling.document_converter import DocumentConverter
    from docling_core.transforms.serializer.markdown import (
        MarkdownDocSerializer,
        MarkdownParams,
        MarkdownTableSerializer,
    )
    from src.extraction_v2.semantic_chunker import AnnotationPictureSerializer

    converter = DocumentConverter()
    result = converter.convert(local_path)
    docling_document = result.document

    # Export per-page markdown using custom serializers
    num_pages = docling_document.num_pages() if hasattr(docling_document, 'num_pages') and callable(docling_document.num_pages) else 0
    pages = []

    if num_pages > 0:
        for page_no in range(1, num_pages + 1):
            try:
                serializer = MarkdownDocSerializer(
                    doc=docling_document,
                    picture_serializer=AnnotationPictureSerializer(),
                    table_serializer=MarkdownTableSerializer(),
                    params=MarkdownParams(pages={page_no}),
                )
                page_md = serializer.serialize().text
                if page_md and page_md.strip():
                    pages.append({"page_number": page_no, "markdown": page_md})
            except Exception as e:
                log.warning(f"Failed to serialize page {page_no}: {e}")
    else:
        # Fallback: export full document
        serializer = MarkdownDocSerializer(
            doc=docling_document,
            picture_serializer=AnnotationPictureSerializer(),
            table_serializer=MarkdownTableSerializer(),
        )
        markdown = serializer.serialize().text
        if markdown:
            pages.append({"page_number": 1, "markdown": markdown})

    if not pages:
        raise ValueError(f"Failed to convert PowerPoint document: {local_path}")

    total_chars = sum(len(p["markdown"]) for p in pages)
    log.info(f"Successfully converted PowerPoint document: {len(pages)} pages, {total_chars} characters")

    # Save large data to pickle file to avoid XCom size limits
    parse_data = {
        'doc_type': 'powerpoint',
        'pages': pages,
        'local_path': local_path,
        'document_id': document_id,
        'original_filename': original_filename,
    }

    data_file = f"/opt/airflow/logs/parse_ppt_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(parse_data, f)

    log.info(f"Saved parse result to {data_file}")

    # Return only small metadata via XCom
    return {
        'data_file': data_file,
        'doc_type': 'powerpoint',
        'document_id': document_id,
        'original_filename': original_filename,
        'local_path': local_path,
        'page_count': len(pages),
        'content_length': total_chars,
    }


@task.external_python(python=EXTERNAL_PYTHON, **_PARSER_RETRIES)
def parse_excel_detailed(
    fetch_result: Dict[str, Any],
    tenant_id: str,
    node_id: str,
    correlation_id: str,
) -> Dict[str, Any]:
    """
    Parse Excel document with DETAILED extraction (images, shapes, flowcharts, comments).

    This pipeline extracts visual elements and detailed content using openpyxl + Docling:
    - Images with AI descriptions
    - Shapes and annotations
    - Flowcharts with workflow descriptions
    - Comments
    - Text content chunked by sheet

    Args:
        fetch_result: Result from fetch_document task containing local_path and document_id
        tenant_id: Tenant UUID for pipeline-event routing
        node_id: Node UUID for pipeline-event routing
        correlation_id: Airflow dag_run_id, for event correlation across stages

    Returns:
        Dict with data_file path and metadata (chunks stored in pickle file)
    """
    # All imports must be inside for external_python
    import sys
    sys.path.insert(0, '/opt/airflow')

    from src.extraction_v2.excel_parser_service import get_excel_parser_service
    from src.extraction_v2.excel_conversion_utils import convert_xls_to_xlsx
    from src.extraction_v2.strikethrough_utils import remove_strikethrough_cells
    from uuid import UUID
    import logging
    import pickle
    import asyncio
    import shutil
    import tempfile
    import os

    log = logging.getLogger(__name__)

    if not fetch_result:
        raise ValueError("No fetch result provided")

    local_path = fetch_result['local_path']
    document_id = fetch_result['document_id']
    original_filename = fetch_result.get('original_filename', 'document.xlsx')

    from src.services.pipeline_events import publish_stage_event
    publish_stage_event(
        "parsing", "started",
        document_id=document_id, tenant_id=tenant_id,
        node_id=node_id, correlation_id=correlation_id,
    )

    log.info(f"Parsing Excel detailed (markdown only): {document_id}, path: {local_path}")

    # Check for password-protected files before processing
    from src.extraction_v2.password_detection import check_password_protected
    check_password_protected(local_path)

    # Copy file to task-specific temp directory to avoid race condition
    # with parse_excel_large_tables running in parallel
    task_temp_dir = tempfile.mkdtemp(prefix=f'parse_detailed_{document_id}_')
    task_local_path = os.path.join(task_temp_dir, os.path.basename(local_path))
    shutil.copy2(local_path, task_local_path)
    log.info(f"Copied file to task-specific path: {task_local_path}")
    local_path = task_local_path

    # Step 1: Convert .xls to .xlsx if needed (preprocessing)
    xlsx_path, was_converted = convert_xls_to_xlsx(local_path)
    if was_converted:
        log.info(f"Converted .xls to .xlsx: {xlsx_path}")
        local_path = xlsx_path

    # Step 2: Remove strikethrough text (preprocessing)
    strikethrough_result = remove_strikethrough_cells(local_path)
    if strikethrough_result['cells_modified'] > 0:
        log.info(
            f"Removed strikethrough text: {strikethrough_result['runs_removed']} runs "
            f"from {strikethrough_result['cells_modified']} cells across "
            f"{strikethrough_result['sheets_affected']} sheets"
        )

    # Use the cleaned file for parsing
    file_to_parse = strikethrough_result['output_file']

    # Read file content
    with open(file_to_parse, 'rb') as f:
        file_content = f.read()

    # Get parser service and parse to per-sheet markdown (no chunking)
    parser_service = get_excel_parser_service()

    parse_result = asyncio.run(
        parser_service.parse_to_markdown(
            file_content=file_content,
            file_id=UUID(document_id),
            original_filename=original_filename,
        )
    )

    sheets = parse_result['sheets']
    image_chunks = parse_result['image_chunks']
    flowchart_chunks = parse_result['flowchart_chunks']

    log.info(
        f"Detailed parse complete: {len(sheets)} sheets, "
        f"{len(image_chunks)} images, {len(flowchart_chunks)} flowcharts"
    )

    # Save parse result to pickle file (per-sheet markdown + image/flowchart DTOs)
    data_file = f"/opt/airflow/logs/parse_detailed_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(parse_result, f)

    log.info(f"Saved detailed parse result to {data_file}")

    # Return metadata via XCom
    return {
        'data_file': data_file,
        'document_id': document_id,
        'filename': original_filename,
        'sheet_count': len(sheets),
        'image_count': len(image_chunks),
        'flowchart_count': len(flowchart_chunks),
    }