"""
Download Routes
Endpoints for downloading Excel files with question responses
"""
from fastapi import APIRouter, HTTPException, Response, Depends, Request
from fastapi.responses import FileResponse
from pydantic import BaseModel, model_validator
from typing import Optional, List, Dict, Any, Union, Tuple
from datetime import datetime
import openpyxl
from openpyxl.drawing.image import Image
from openpyxl.styles import Font, Alignment, PatternFill
from openpyxl.utils import get_column_letter
import tempfile
import plotly.graph_objects as go
import plotly.express as px
import io
import pandas as pd
import os
import ast
import json
import uuid
import re
import html as html_module
from urllib.parse import parse_qs, unquote, urlparse
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt



from app.core.config import settings
from app.services.askme_service import (
    get_askme_service,
)
from app.services.agent_config import get_agent_config
from app.services.user_question_log_cache import (
    canonical_agent_id_for_cache,
    find_reusable_log_row,
    normalize_question_for_cache,
)
from app.core.dependencies import get_current_user
from app.models.user import User
from app.models.question import Question
from app.core.database import get_db
from sqlalchemy import text
from sqlalchemy.orm import Session
from fastapi.concurrency import run_in_threadpool



router = APIRouter(prefix="/download", tags=["Download"])


# Schemas
class DocumentEmailRequest(BaseModel):
    """Unified email request for @file (document_search) and @find (document_find) responses."""
    agent_id: str
    sendemail: str  # "Y" or "N"
    email_list: Optional[List[str]] = None
    my_question: Optional[str] = None
    question: Optional[str] = None
    document_result_type: Optional[str] = None  # document_search | document_find
    is_file: Optional[bool] = None
    # @file — one chunk from document_search.chunk_hits
    chunk_id: Optional[str] = None
    filename: Optional[str] = None
    relevant_images: Optional[List[Dict[str, Any]]] = None
    document_search: Optional[Dict[str, Any]] = None
    # @find — one file from document_find.results
    file_id: Optional[str] = None
    page_images: Optional[Dict[str, List[Dict[str, Any]]]] = None
    document_find: Optional[Dict[str, Any]] = None

    @model_validator(mode="after")
    def _resolve_question(self):
        if not (self.my_question or "").strip() and (self.question or "").strip():
            self.my_question = str(self.question).strip()
        return self


# Backward-compatible alias
DocumentChunkEmailRequest = DocumentEmailRequest


class DownloadExcelRequest(BaseModel):
    """Request model for downloading Excel file.
    agent_id accepts a single string (single-agent) or a list of strings (multi-agent).
    The /download-excel endpoint uses only single-agent (first item is extracted).
    The /multi-agent-download-excel endpoint uses the full list.
    """
    my_question: Optional[str] = None
    question: Optional[str] = None
    agent_id: Union[str, List[str]]  # Single agent ID string OR list for multi-agent
    sendemail: str  # "Y" or "N"
    email_list: Optional[List[str]] = None
    sql: Optional[Union[str, List[str]]] = None
    code: Optional[str] = None  # Chart code
    api_endpoint: Optional[Union[str, List[str]]] = None
    api_params: Optional[Union[str, List[str]]] = None
    api_response_validator: Optional[Union[str, List[str]]] = None
    question_id: Optional[str] = None
    # Document email fields (@file / @find)
    is_file: Optional[bool] = None
    chunk_id: Optional[str] = None
    file_id: Optional[str] = None
    relevant_images: Optional[List[Dict[str, Any]]] = None
    page_images: Optional[Dict[str, List[Dict[str, Any]]]] = None
    document_result_type: Optional[str] = None
    document_search: Optional[Dict[str, Any]] = None
    document_find: Optional[Dict[str, Any]] = None
    filename: Optional[str] = None
    # Prior table for @askme / optional for @web — list of row dicts, {headers,data}, JSON string, or omit
    # if sql / api_* / question_id can supply data (see _resolve_prior_df_input_for_askme).
    df: Optional[Any] = None
    # Optional: reuse prior @askme ask-response (skip re-analysis when exporting to Excel)
    text_analysis: Optional[str] = None
    text_analysis_html: Optional[str] = None 
    # Optional: SQL summary blocks from ask-response/details
    sql_query_summaries: Optional[Any] = None

    @model_validator(mode="after")
    def _resolve_question(self):
        if not (self.my_question or "").strip() and (self.question or "").strip():
            self.my_question = str(self.question).strip()
        if not (self.my_question or "").strip():
            raise ValueError("Either my_question or question is required")
        return self

def parse_api_field(field_value: Union[str, List[str]], default=None) -> List[str]:
    """
    Helper to robustly parse API fields that might be strings, list of strings, 
    or JSON-encoded lists.
    """
    if not field_value or field_value == "N":
        return default or []
    
    if isinstance(field_value, list):
        return field_value
        
    if isinstance(field_value, str):
        field_value = field_value.strip()
        if field_value.startswith('[') and field_value.endswith(']'):
            try:
                # Use standard json module locally
                import json
                parsed = json.loads(field_value)
                if isinstance(parsed, list):
                    return [str(item) for item in parsed]
            except Exception:
                pass
        return [field_value]
    
    return [str(field_value)] if field_value else []


def _is_document_question(
    question: Optional[str],
    document_result_type: Optional[str] = None,
    is_file: Optional[bool] = None,
) -> bool:
    """True for @file / @find / @analyze, explicit document types, or is_file flag."""
    if is_file:
        return True
    if document_result_type in ("document_search", "document_analysis", "document_find"):
        return True
    q = (question or "").lower()
    return any(token in q for token in ("@file", "@find", "@analyze", "@analyse"))


def _resolve_download_question_text(request: DownloadExcelRequest) -> str:
    return str(getattr(request, "my_question", None) or getattr(request, "question", None) or "").strip()


def _page_sort_key(page_key: Any) -> Any:
    try:
        return int(page_key)
    except (TypeError, ValueError):
        return str(page_key)


def _coerce_page_images(value: Any) -> Optional[Dict[str, List[Dict[str, Any]]]]:
    """Accept page_images as dict or JSON string; normalize page keys to str."""
    if value is None:
        return None
    parsed: Any = value
    if isinstance(parsed, str):
        s = parsed.strip()
        if not s:
            return None
        try:
            parsed = json.loads(s)
        except Exception:
            return None
    if not isinstance(parsed, dict):
        return None
    out: Dict[str, List[Dict[str, Any]]] = {}
    for page_key, imgs in parsed.items():
        if isinstance(imgs, list):
            out[str(page_key)] = [img for img in imgs if isinstance(img, dict)]
        elif isinstance(imgs, dict):
            out[str(page_key)] = [imgs]
    return out or None


def _append_page_citation(
    images: List[Dict[str, Any]],
    *,
    page: Any,
    page_download_url: Optional[str] = None,
    file_id: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> None:
    if page is None:
        return
    try:
        page_n = int(page)
    except (TypeError, ValueError):
        return
    if page_n < 1:
        return
    row: Dict[str, Any] = {"page": page_n}
    if page_download_url:
        row["path"] = str(page_download_url).strip()
    if file_id:
        row["file_id"] = str(file_id).strip()
    if agent_id is not None and str(agent_id).strip():
        row["agent_id"] = str(agent_id).strip()
    images.append(row)


def _normalize_document_email_images(
    *,
    relevant_images: Optional[List[Dict[str, Any]]] = None,
    page_images: Optional[Any] = None,
    matching_page_urls: Optional[List[Dict[str, Any]]] = None,
) -> List[Dict[str, Any]]:
    """Normalize legacy images and page_download_url citations for document email."""
    images: List[Dict[str, Any]] = []
    coerced_pages = _coerce_page_images(page_images)

    for img in relevant_images or []:
        if not isinstance(img, dict):
            continue
        if img.get("page_download_url"):
            _append_page_citation(
                images,
                page=img.get("page"),
                page_download_url=img.get("page_download_url"),
                file_id=img.get("file_id"),
                agent_id=img.get("agent_id"),
            )
        elif img.get("path"):
            images.append(img)
        elif img.get("page") is not None:
            images.append(dict(img))

    for row in matching_page_urls or []:
        if isinstance(row, dict):
            if row.get("path"):
                entry = dict(row)
                if entry.get("page") is None and entry.get("page_number") is not None:
                    entry["page"] = entry.get("page_number")
                images.append(entry)
            else:
                _append_page_citation(
                    images,
                    page=row.get("page"),
                    page_download_url=row.get("page_download_url"),
                    file_id=row.get("file_id"),
                    agent_id=row.get("agent_id"),
                )

    if coerced_pages:
        for page_key in sorted(coerced_pages.keys(), key=_page_sort_key):
            for img in coerced_pages.get(page_key) or []:
                if isinstance(img, dict) and img.get("path"):
                    entry = img.copy()
                    if entry.get("page") is None and str(page_key).isdigit():
                        entry["page"] = int(page_key)
                    images.append(entry)

    seen = set()
    unique: List[Dict[str, Any]] = []
    for img in images:
        key = (
            str(img.get("file_id") or ""),
            str(img.get("agent_id") or ""),
            str(img.get("page") or ""),
            str(img.get("path") or ""),
        )
        if key in seen:
            continue
        if key[3] or key[2] or (key[0] and key[2]):
            seen.add(key)
            unique.append(img)
    unique.sort(key=lambda x: (x.get("page") is None, x.get("page") or 0))
    return unique


def _infer_document_result_type(
    question: Optional[str],
    explicit: Optional[str] = None,
) -> Optional[str]:
    if explicit:
        return explicit
    q = (question or "").lower()
    if "@find" in q:
        return "document_find"
    if "@file" in q:
        return "document_search"
    return None


def _extract_find_search_term(my_question: Optional[str]) -> Optional[str]:
    q = (my_question or "").strip()
    if q.lower().startswith("@find"):
        return q[5:].strip() or None
    return None


def _find_result_matches_target(
    result: Dict[str, Any],
    *,
    filename: Optional[str] = None,
    file_id: Optional[str] = None,
) -> bool:
    if file_id and str(result.get("file_id") or "") != str(file_id):
        return False
    if filename:
        target = str(filename).strip().lower()
        result_name = str(result.get("filename") or "").strip().lower()
        if target and result_name and target != result_name:
            return False
    return True


def _run_find_query_for_email(agent_id: str, search_query: str, agent_config: dict) -> Optional[Dict[str, Any]]:
    try:
        from app.services.document_service import (
            DocumentService,
            FILE_EXTRACT_DISK_ROOT,
            FILE_IMAGE_DISK_ROOT,
            FILE_SYS_METADATA_DB,
            FILE_UPLOAD_DISK_ROOT,
            FILE_VECTORS_DB_DIR,
        )

        api_key = agent_config.get("api_key") or agent_config.get("open_ai")
        if not api_key:
            return None
        doc = DocumentService(
            upload_folder=FILE_UPLOAD_DISK_ROOT,
            image_folder=FILE_IMAGE_DISK_ROOT,
            extracted_folder=FILE_EXTRACT_DISK_ROOT,
            db_file=FILE_SYS_METADATA_DB,
            chroma_path=FILE_VECTORS_DB_DIR,
            openai_api_key=str(api_key),
            claude_api_key=str(agent_config.get("claude_api_key") or ""),
            agent_id=agent_id,
        )
        return doc.handle_query(f"@find {search_query.strip()}")
    except Exception as exc:
        print(f"[SEND-DOCUMENT-EMAIL] find re-query error: {exc}")
        return None


def _lookup_find_email_pages(
    agent_id: str,
    filename: Optional[str] = None,
    file_id: Optional[str] = None,
    search_query: Optional[str] = None,
    *,
    agent_config: Optional[dict] = None,
) -> Tuple[Optional[Dict[str, List[Dict[str, Any]]]], Optional[str], Optional[str]]:
    """
    Resolve @find page_images when the payload only has filename/file_id.
    Re-runs @find when possible; builds runtime page-image API paths (no extracted_images).
    """
    if not filename and not file_id:
        return None, None, None
    try:
        from app.services.document_db_service import DocumentDbService
        from app.utils.document_preview import build_page_images_from_api_paths

        db_svc = DocumentDbService(db_file="file_sys_metadata.db", agent_id=agent_id)
        try:
            row = None
            if file_id:
                row = db_svc.get_file(file_id, agent_id=agent_id)
            if not row and filename:
                name = str(filename).strip()
                query = (
                    "SELECT id, filename, original_filename, file_type, file_path, "
                    "folder_path, relative_path, upload_date, file_size, content_preview, "
                    "full_content, image_path, thumbnail_path, agent_id "
                    "FROM files WHERE LOWER(original_filename) = LOWER(:name) "
                    "OR LOWER(filename) = LOWER(:name)"
                )
                params: Dict[str, Any] = {"name": name}
                query = db_svc._add_agent_filter(query, params, agent_id)
                query += " ORDER BY upload_date DESC LIMIT 1"
                result = db_svc.db.execute(text(query), params).fetchone()
                row = tuple(result) if result else None

            resolved_file_id = str(row[0]) if row else (str(file_id).strip() if file_id else None)
            resolved_filename = (row[2] or row[1]) if row else filename
            resolved_agent = (
                str(row[13]) if row and len(row) > 13 and row[13] else str(agent_id)
            )

            if search_query and agent_config:
                find_result = _run_find_query_for_email(agent_id, search_query, agent_config)
                if isinstance(find_result, dict) and find_result.get("type") == "document_find":
                    for result in find_result.get("results") or []:
                        if not isinstance(result, dict):
                            continue
                        if not _find_result_matches_target(
                            result, filename=filename, file_id=resolved_file_id
                        ):
                            continue
                        result_pages = result.get("page_images")
                        if isinstance(result_pages, dict) and result_pages:
                            return (
                                _coerce_page_images(result_pages),
                                result.get("filename") or resolved_filename,
                                result.get("file_id") or resolved_file_id,
                            )
                        matching_pages = result.get("matching_pages") or []
                        fid = str(result.get("file_id") or resolved_file_id or "")
                        aid = str(result.get("agent_id") or resolved_agent or agent_id)
                        if fid and matching_pages:
                            return (
                                build_page_images_from_api_paths(fid, aid, matching_pages),
                                result.get("filename") or resolved_filename,
                                fid,
                            )

            return None, resolved_filename, resolved_file_id
        finally:
            db_svc.close()
    except Exception as exc:
        print(f"[SEND-DOCUMENT-EMAIL] find email page lookup error: {exc}")
        return None, None, None


def _lookup_page_images_from_document_db(
    agent_id: str,
    filename: Optional[str] = None,
    file_id: Optional[str] = None,
    search_query: Optional[str] = None,
    *,
    agent_config: Optional[dict] = None,
) -> Tuple[Optional[Dict[str, List[Dict[str, Any]]]], Optional[str], Optional[str]]:
    """Backward-compatible alias for @find email page resolution."""
    return _lookup_find_email_pages(
        agent_id,
        filename,
        file_id,
        search_query,
        agent_config=agent_config,
    )


async def _hydrate_document_email_images_from_db(
    *,
    agent_id: str,
    my_question: Optional[str],
    document_result_type: Optional[str],
    relevant_images: Optional[List[Dict[str, Any]]],
    page_images: Optional[Any],
    document_search: Optional[Dict[str, Any]],
    document_find: Optional[Dict[str, Any]],
    chunk_id: Optional[str],
    file_id: Optional[str],
    filename: Optional[str],
    agent_config: Optional[dict] = None,
) -> Tuple[
    Optional[str],
    Optional[List[Dict[str, Any]]],
    Optional[Any],
    Optional[str],
    Optional[str],
    Optional[str],
]:
    """
    Fill missing @find page_images from document DB (filename/file_id only).

    @file (document_search) is unchanged: uses relevant_images + chunk_id from the payload
    or document_search block — no DB fallback.
    """
    effective_type = _infer_document_result_type(my_question, document_result_type)
    question_lower = (my_question or "").lower()
    is_find = effective_type == "document_find" or (
        effective_type is None and "@find" in question_lower
    )
    is_file = effective_type == "document_search" or (
        effective_type is None and "@file" in question_lower and "@find" not in question_lower
    )

    if is_file:
        return (
            effective_type or "document_search",
            relevant_images,
            page_images,
            chunk_id,
            file_id,
            filename,
        )

    if not is_find:
        return effective_type, relevant_images, page_images, chunk_id, file_id, filename

    has_find_images = bool(
        _normalize_document_email_images(page_images=_coerce_page_images(page_images))
        or document_find
    )
    if has_find_images:
        return effective_type or "document_find", relevant_images, page_images, chunk_id, file_id, filename

    looked_up_pages, looked_up_filename, looked_up_file_id = await run_in_threadpool(
        _lookup_find_email_pages,
        agent_id,
        filename,
        file_id,
        _extract_find_search_term(my_question),
        agent_config=agent_config,
    )
    if not looked_up_pages and relevant_images:
        fallback_pages: List[int] = []
        for img in relevant_images:
            if not isinstance(img, dict) or img.get("page") is None:
                continue
            try:
                fallback_pages.append(int(img["page"]))
            except (TypeError, ValueError):
                continue
        if fallback_pages:
            try:
                from app.utils.document_preview import build_page_images_from_api_paths

                _, resolved_filename, resolved_file_id = await run_in_threadpool(
                    _lookup_find_email_pages,
                    agent_id,
                    filename,
                    file_id,
                    None,
                    agent_config=agent_config,
                )
                if resolved_file_id:
                    looked_up_pages = build_page_images_from_api_paths(
                        str(resolved_file_id),
                        str(agent_id),
                        fallback_pages,
                    )
                    filename = filename or resolved_filename
                    file_id = file_id or resolved_file_id
            except Exception as exc:
                print(f"[SEND-DOCUMENT-EMAIL] find page fallback error: {exc}")
    if not looked_up_pages:
        return effective_type or "document_find", relevant_images, page_images, chunk_id, file_id, filename

    page_images = looked_up_pages
    filename = filename or looked_up_filename
    file_id = file_id or looked_up_file_id
    return "document_find", relevant_images, page_images, chunk_id, file_id, filename


def _resolve_document_email_payload(
    *,
    document_result_type: Optional[str] = None,
    relevant_images: Optional[List[Dict[str, Any]]] = None,
    page_images: Optional[Dict[str, List[Dict[str, Any]]]] = None,
    document_search: Optional[Dict[str, Any]] = None,
    document_find: Optional[Dict[str, Any]] = None,
    chunk_id: Optional[str] = None,
    file_id: Optional[str] = None,
    filename: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[str], Optional[str]]:
    """
    Resolve images + metadata for document email from @file or @find payloads.

    Returns (images, filename, chunk_id, file_id).
    """
    coerced_pages = _coerce_page_images(page_images)

    if document_result_type == "document_find":
        find_page_urls: Optional[List[Dict[str, Any]]] = None
        if isinstance(document_find, dict):
            for result in document_find.get("results") or []:
                if isinstance(result, dict) and result.get("matching_page_urls"):
                    find_page_urls = result.get("matching_page_urls")
                    break
        images = _normalize_document_email_images(
            relevant_images=relevant_images,
            page_images=coerced_pages,
            matching_page_urls=find_page_urls,
        )
    elif document_result_type == "document_search":
        images = _normalize_document_email_images(relevant_images=relevant_images)
    else:
        images = _normalize_document_email_images(
            relevant_images=relevant_images,
            page_images=coerced_pages,
        )
    resolved_filename = filename
    resolved_chunk_id = chunk_id
    resolved_file_id = file_id

    if not images and isinstance(document_search, dict):
        for hit in document_search.get("chunk_hits") or []:
            if not isinstance(hit, dict):
                continue
            if chunk_id and hit.get("chunk_id") != chunk_id:
                continue
            if hit.get("page_download_url") and hit.get("page_number") is not None:
                images = _normalize_document_email_images(
                    relevant_images=[
                        {
                            "page": hit.get("page_number"),
                            "page_download_url": hit.get("page_download_url"),
                            "file_id": hit.get("file_id"),
                            "agent_id": hit.get("agent_id"),
                        }
                    ],
                )
                resolved_filename = resolved_filename or hit.get("filename")
                resolved_chunk_id = resolved_chunk_id or hit.get("chunk_id")
                resolved_file_id = resolved_file_id or hit.get("file_id")
                break
            if hit.get("path") and hit.get("page_number") is not None:
                images = _normalize_document_email_images(
                    relevant_images=[
                        {
                            "page": hit.get("page_number"),
                            "path": hit.get("path"),
                            "file_id": hit.get("file_id"),
                            "agent_id": hit.get("agent_id"),
                        }
                    ],
                )
                resolved_filename = resolved_filename or hit.get("filename")
                resolved_chunk_id = resolved_chunk_id or hit.get("chunk_id")
                resolved_file_id = resolved_file_id or hit.get("file_id")
                break
            hit_images = hit.get("relevant_images") or []
            if hit_images:
                images = _normalize_document_email_images(relevant_images=hit_images)
                resolved_filename = resolved_filename or hit.get("filename")
                resolved_chunk_id = resolved_chunk_id or hit.get("chunk_id")
                break
        if not images and not chunk_id:
            for hit in document_search.get("chunk_hits") or []:
                if not isinstance(hit, dict):
                    continue
                if hit.get("page_download_url") and hit.get("page_number") is not None:
                    images = _normalize_document_email_images(
                        relevant_images=[
                            {
                                "page": hit.get("page_number"),
                                "page_download_url": hit.get("page_download_url"),
                                "file_id": hit.get("file_id"),
                                "agent_id": hit.get("agent_id"),
                            }
                        ],
                    )
                    resolved_filename = resolved_filename or hit.get("filename")
                    resolved_chunk_id = resolved_chunk_id or hit.get("chunk_id")
                    resolved_file_id = resolved_file_id or hit.get("file_id")
                    break

    if not images and isinstance(document_find, dict):
        for result in document_find.get("results") or []:
            if not isinstance(result, dict):
                continue
            if file_id and result.get("file_id") != file_id:
                continue
            if filename and not file_id:
                if not _find_result_matches_target(result, filename=filename):
                    continue
            page_urls = result.get("matching_page_urls") or []
            if page_urls:
                enriched = []
                for row in page_urls:
                    if not isinstance(row, dict):
                        continue
                    enriched.append(
                        {
                            **row,
                            "file_id": row.get("file_id") or result.get("file_id"),
                            "agent_id": row.get("agent_id") or result.get("agent_id"),
                        }
                    )
                images = _normalize_document_email_images(matching_page_urls=enriched)
                resolved_filename = resolved_filename or result.get("filename")
                resolved_file_id = resolved_file_id or result.get("file_id")
                break
            result_pages = result.get("page_images")
            if isinstance(result_pages, dict) and result_pages:
                images = _normalize_document_email_images(
                    page_images=_coerce_page_images(result_pages),
                )
                resolved_filename = resolved_filename or result.get("filename")
                resolved_file_id = resolved_file_id or result.get("file_id")
                break
        if not images and not file_id:
            for result in document_find.get("results") or []:
                if not isinstance(result, dict):
                    continue
                result_pages = result.get("page_images")
                if isinstance(result_pages, dict) and result_pages:
                    images = _normalize_document_email_images(
                        page_images=_coerce_page_images(result_pages),
                    )
                    resolved_filename = resolved_filename or result.get("filename")
                    resolved_file_id = resolved_file_id or result.get("file_id")
                    break

    if document_result_type == "document_search" and not resolved_chunk_id and images:
        resolved_chunk_id = chunk_id
    if not resolved_file_id and chunk_id:
        chunk_fid, _ = _file_id_page_from_chunk_id(chunk_id)
        resolved_file_id = resolved_file_id or chunk_fid

    return images, resolved_filename, resolved_chunk_id, resolved_file_id


def _trim_trailing_empty_api_slots(
    endpoints: List[str], params: List[str], validators: List[str]
) -> Tuple[List[str], List[str], List[str]]:
    """Drop trailing blank slots from JSON arrays (e.g. ['url',''] mis-sent from UI)."""
    e, p, v = list(endpoints or []), list(params or []), list(validators or [])
    while e and not str(e[-1]).strip():
        e.pop()
        if p:
            p.pop()
        if v:
            v.pop()
    return e, p, v


def _hydrate_request_from_user_question_log_cache(
    request: DownloadExcelRequest,
    user_id: str,
) -> bool:
    """Hydrate SQL/API fields from latest successful user_question_log for same question."""
    question = _resolve_download_question_text(request)
    if not question:
        return False
    if _is_document_question(
        question,
        getattr(request, "document_result_type", None),
        getattr(request, "is_file", None),
    ):
        return False
    qn = normalize_question_for_cache(question)
    if not qn:
        return False
    row = find_reusable_log_row(question_normalized=qn, user_id=user_id)
    if row is None:
        return False
    if canonical_agent_id_for_cache(getattr(request, "agent_id", None)) != canonical_agent_id_for_cache(
        getattr(row, "agent_id", None)
    ):
        return False

    hydrated = False
    if not getattr(request, "sql", None) and getattr(row, "sql", None):
        request.sql = row.sql
        hydrated = True
    if not getattr(request, "api_endpoint", None) and getattr(row, "api_endpoint", None):
        request.api_endpoint = row.api_endpoint
        hydrated = True
    if not getattr(request, "api_params", None) and getattr(row, "api_params", None):
        request.api_params = row.api_params
        hydrated = True
    if (
        not getattr(request, "api_response_validator", None)
        and getattr(row, "api_response_validator", None)
    ):
        request.api_response_validator = row.api_response_validator
        hydrated = True
    if (
        not getattr(request, "sql_query_summaries", None)
        and getattr(row, "sql_query_summaries", None)
    ):
        request.sql_query_summaries = row.sql_query_summaries
        hydrated = True
    return hydrated


def _download_meaningful_sql(sql_val: Any) -> bool:
    """Aligned with multi-agent-ask: N / empty means no SQL to reuse."""
    if sql_val is None:
        return False
    if isinstance(sql_val, str):
        s = sql_val.strip().upper()
        return bool(s and s not in ("N", "NONE", "[]", "{}"))
    if isinstance(sql_val, (dict, list)):
        return bool(sql_val)
    return True


def _download_meaningful_api(api_endpoint: Any, api_params: Any) -> bool:
    if api_endpoint is None or api_params is None:
        return False
    if isinstance(api_endpoint, str) and api_endpoint.strip().upper() in ("", "N", "NONE", "[]"):
        return False
    if isinstance(api_params, str) and api_params.strip().upper() in ("", "N", "NONE", "[]"):
        return False
    return True


def _download_merge_info_from_sql_blob(blob: Any) -> Optional[Dict[str, Any]]:
    """
    Normalize saved/request multi-agent SQL payload into merge_info.

    Supports dict/string ``{queries...}``, dict/string ``{agent_results...}``,
    and new list format ``[{agent_id, sub_question, sql}, ...]``.
    """
    parsed: Any = blob
    if isinstance(parsed, str):
        s = parsed.strip()
        if not s.startswith(("{", "[")):
            return None
        try:
            parsed = json.loads(s)
        except Exception:
            return None

    if isinstance(parsed, dict):
        if "queries" in parsed and isinstance(parsed.get("queries"), dict):
            queries = {
                str(k): (v if isinstance(v, str) else json.dumps(v, default=str))
                for k, v in (parsed.get("queries") or {}).items()
                if str(k).strip() and str(v if isinstance(v, str) else json.dumps(v, default=str)).strip()
            }
            if not queries:
                return None
            order = parsed.get("execution_order") or list(queries.keys())
            execution_order = [
                str(x).strip()
                for x in (order if isinstance(order, list) else list(queries.keys()))
                if str(x).strip() in queries
            ] or list(queries.keys())
            return {
                "queries": queries,
                "execution_order": execution_order,
                "merge_key": parsed.get("merge_key", "id"),
                "merge_strategy": parsed.get("merge_strategy", "independent"),
            }
        if "agent_results" in parsed:
            parsed = parsed.get("agent_results") or []
        else:
            return None

    if isinstance(parsed, list):
        queries: Dict[str, str] = {}
        execution_order: List[str] = []
        for idx, item in enumerate(parsed):
            if not isinstance(item, dict):
                continue
            aid = str(
                item.get("agent_id")
                or item.get("id")
                or item.get("agent")
                or f"agent{idx + 1}"
            ).strip()
            sql_part = item.get("sql")
            if sql_part is None:
                continue
            sql_text = sql_part if isinstance(sql_part, str) else json.dumps(sql_part, default=str)
            if not str(sql_text).strip():
                continue
            queries[aid] = str(sql_text)
            if aid not in execution_order:
                execution_order.append(aid)
        if not queries:
            return None
        return {
            "queries": queries,
            "execution_order": execution_order or list(queries.keys()),
            "merge_key": "id",
            "merge_strategy": "independent",
        }
    return None


async def _download_fetch_dataframe_from_saved_api(
    service: Any,
    request: DownloadExcelRequest,
    single_agent_id: str,
) -> Tuple[Optional[pd.DataFrame], Optional[str]]:
    """
    Run the same saved API path as /download-excel (single or merged multi-API) and return a DataFrame.
    Used to hydrate prior rows for @askme when 'df' is omitted.
    """
    api_endpoints = parse_api_field(request.api_endpoint)
    api_params_list = parse_api_field(request.api_params)
    api_validators = parse_api_field(request.api_response_validator)
    if not api_endpoints or not _download_meaningful_api(
        request.api_endpoint, request.api_params
    ):
        return None, None

    try:
        if len(api_endpoints) > 1:
            all_apis = await run_in_threadpool(service.get_all_apis, single_agent_id)
            selected_apis = []
            for ep in api_endpoints:
                for api in all_apis:
                    if api.get("endpoint") == ep:
                        selected_apis.append(api)
                        break
            if not selected_apis:
                return None, f"API endpoints {api_endpoints} not found for agent '{single_agent_id}'"
            svc_api = service._ensure_api_chat_service()
            merge_result = await run_in_threadpool(
                svc_api.call_multiple_apis_and_merge,
                selected_apis=selected_apis,
                user_query=request.my_question,
            )
            if not merge_result.get("success"):
                return None, merge_result.get(
                    "message", "Failed to retrieve data from multiple APIs"
                )
            return merge_result.get("df"), None

        api_endpoint_str = api_endpoints[0]
        api_params_str = api_params_list[0] if api_params_list else ""
        api_response_validator_str = api_validators[0] if api_validators else ""

        try:
            extracted_params = (
                ast.literal_eval(api_params_str)
                if isinstance(api_params_str, str) and api_params_str.strip()
                else {}
            )
            if not isinstance(extracted_params, dict):
                extracted_params = {}
        except Exception:
            extracted_params = {}

        apis_list = await run_in_threadpool(service.get_all_apis, single_agent_id)
        api_info = {}
        for api in apis_list:
            if api.get("endpoint") == api_endpoint_str:
                api_info = api
                break

        if not api_info:
            return (
                None,
                f"API endpoint '{api_endpoint_str}' not found for agent '{single_agent_id}'",
            )

        is_sap = api_info.get("extra1", "N")
        if is_sap == "Y":
            sap_result = await run_in_threadpool(
                service.process_sap_query,
                service_root=api_endpoint_str,
                entity_set=api_info.get("base_params", ""),
                user_query=request.my_question,
                response_schema=api_info.get("response_schema", ""),
                transformation_code=api_response_validator_str or "",
                username="S4h_aa",
                password="Welcome1",
                verify_ssl=False,
                api_query_params=extracted_params,
            )
            return sap_result.get("df"), None

        _data, _data2, df, _data3 = await run_in_threadpool(
            service.call_api_and_transform,
            extracted_params,
            api_endpoint_str,
            api_info.get("response_schema", ""),
            api_response_validator_str or "",
            request.my_question,
        )
        return df, None
    except Exception as e:
        return None, f"Error calling API for prior data: {e}"


def _normalize_sql_string_for_download(request: DownloadExcelRequest) -> Optional[str]:
    sql_list = parse_api_field(request.sql)
    sql_str = sql_list[0] if sql_list else None
    if sql_str and isinstance(sql_str, str):
        sql_str = sql_str.strip() or None
    if not sql_str or sql_str == "N":
        return None
    return sql_str


async def _resolve_prior_df_input_for_askme(
    service: Any,
    request: DownloadExcelRequest,
    single_agent_id: str,
) -> Tuple[Any, Optional[str]]:
    """
    Resolve tabular input for @askme: request.df first, else run saved SQL, else saved API.
    Returns (df_input suitable for _parse_df_input, error_message).
    """
    parsed = service._parse_df_input(getattr(request, "df", None))
    if parsed and len(parsed) > 0:
        return getattr(request, "df", None), None

    sql_str = _normalize_sql_string_for_download(request)
    if sql_str:
        def _run_sql():
            service._ensure_sql()
            if not service.is_sql_valid_cached(sql=sql_str):
                raise ValueError("Saved SQL is not valid")
            return service.run_sql_cached(sql=sql_str)

        try:
            prior_df = await run_in_threadpool(_run_sql)
        except ValueError as ve:
            return None, str(ve)
        except Exception as e:
            return None, str(e)

        if prior_df is None or getattr(prior_df, "empty", True):
            return (
                None,
                "No rows returned from saved SQL. Provide 'df' or fix the SQL.",
            )
        return prior_df.to_dict("records"), None

    api_df, api_err = await _download_fetch_dataframe_from_saved_api(
        service, request, single_agent_id
    )
    if api_err:
        return None, api_err
    if api_df is not None and not api_df.empty:
        return api_df.to_dict("records"), None

    return None, None


def _prior_input_from_executed_df(df: Any) -> Any:
    """
    Build prior-tabular input for @askme Excel from the df already produced by SQL/API branches.
    Returns row records (list of dicts) suitable for AskmeService._parse_df_input, or None.
    For list[DataFrame] (multi-table), uses the first non-empty frame (same pragmatic choice as chart path).
    """
    if df is None:
        return None
    if isinstance(df, pd.DataFrame):
        if getattr(df, "empty", True):
            return None
        return df.to_dict("records")
    if isinstance(df, list):
        for item in df:
            if isinstance(item, pd.DataFrame) and not getattr(item, "empty", True):
                return item.to_dict("records")
        return None
    return None


def _strip_web_token(q: str) -> str:
    return re.sub(r"@web\b", "", q or "", flags=re.IGNORECASE).strip()


def _strip_askme_token(q: str) -> str:
    return re.sub(r"@askme\b", "", q or "", flags=re.IGNORECASE).strip()


def _parse_web_analysis_text(analysis_result: Any) -> Tuple[str, bool]:
    if isinstance(analysis_result, str):
        t = analysis_result
        return t, bool(t)
    if isinstance(analysis_result, dict):
        t = analysis_result.get("text", analysis_result.get("formatted_response", ""))
        has_text = analysis_result.get("has_text", bool(t))
        return (str(t) if t else ""), bool(has_text and t)
    s = str(analysis_result)
    return s, bool(s)


async def _web_excel_payload(
    service: Any,
    my_question: str,
    df_input: Any,
    code: Optional[str],
) -> Tuple[Optional[str], Optional[pd.DataFrame], Optional[Dict[str, Any]], Optional[str]]:
    """
    Same core behavior as ask-question @web: search_and_format_response + optional prior df + chart from code.
    Returns (summary, df_display, fig, error_message).
    """
    query = _strip_web_token(my_question)
    if not query:
        return None, None, None, "Please provide your search request after @web"

    parsed = service._parse_df_input(df_input)
    analysis_df = pd.DataFrame(parsed) if parsed and len(parsed) > 0 else pd.DataFrame()

    analysis_result = await run_in_threadpool(
        service.search_and_format_response,
        user_query=query,
        last_response=analysis_df,
        use_qwen=settings.default_use_qwen_for_summary_and_web(),
    )
    analysis_text, has_text = _parse_web_analysis_text(analysis_result)
    if not has_text or not (analysis_text or "").strip():
        return None, None, None, "Could not generate web search results. Please try rephrasing your request."

    df_display = None
    fig = None
    if analysis_df is not None and len(analysis_df) > 0:
        df_display = analysis_df.copy()
        df_display.index = df_display.index + 1
        if len(analysis_df) <= 100 and code and str(code).strip() not in ("", "N"):
            try:
                cols = list(analysis_df.columns)
                fig = generate_chart_from_code(code, analysis_df, cols, my_question, service)
            except Exception as e:
                print(f"[DOWNLOAD-EXCEL] @web chart error: {e}")

    return analysis_text, df_display, fig, None


def _plain_text_from_cached_askme(
    text_analysis: Optional[str], text_analysis_html: Optional[str]
) -> str:
    t = (text_analysis or "").strip()
    if t:
        return t
    raw = (text_analysis_html or "").strip()
    if not raw:
        return ""
    plain = re.sub(r"<[^>]+>", " ", raw)
    return html_module.unescape(plain).strip()


def _plain_text_from_sql_query_summaries(sql_query_summaries: Any) -> str:
    """
    Convert sql_query_summaries payload to plain text for Excel summary sheet.
    Accepts list/dict/json-string and returns a compact, human-readable summary.
    """
    if sql_query_summaries is None:
        return ""
    parsed = sql_query_summaries
    if isinstance(parsed, str):
        s = parsed.strip()
        if not s:
            return ""
        if s.startswith("[") or s.startswith("{"):
            try:
                parsed = json.loads(s)
            except Exception:
                return s
        else:
            return s
    if isinstance(parsed, dict):
        parsed = [parsed]
    if not isinstance(parsed, list):
        return str(parsed).strip()

    _agent_name_cache: Dict[str, str] = {}

    def _resolve_agent_name(agent_id_val: Any) -> str:
        aid = str(agent_id_val or "").strip()
        if not aid:
            return "Unknown"
        if aid in _agent_name_cache:
            return _agent_name_cache[aid]
        try:
            conf = get_agent_config(agent_id=aid) or {}
            name = str(conf.get("agent_name") or "").strip()
            if not name:
                name = aid
        except Exception:
            name = aid
        _agent_name_cache[aid] = name
        return name

    lines: List[str] = []
    for item in parsed:
        if not isinstance(item, dict):
            continue
        agent_name = str(item.get("agent_name") or "").strip()
        if not agent_name:
            agent_name = _resolve_agent_name(item.get("agent_id"))
        if not agent_name:
            agent_name = "Unknown"
        sub_q = str(item.get("sub_question") or "").strip()
        summ = str(item.get("summary") or "").strip()
        if not (sub_q or summ):
            continue
        lines.append(f"Agent Name: {agent_name}")
        if sub_q:
            lines.append(f"Sub Question: {sub_q}")
        if summ:
            lines.append(f"Summary: {summ}")
        lines.append("")
    while lines and not lines[-1].strip():
        lines.pop()
    return "\n".join(lines).strip()


def _askme_excel_from_cached_ask_response(
    service: Any,
    request: DownloadExcelRequest,
    prior_input: Any,
) -> Tuple[Optional[str], Optional[pd.DataFrame], Optional[Dict[str, Any]], Optional[str]]:
    """Build Excel parts from prior ask-question @askme payload (no second LLM call)."""
    summary = _plain_text_from_sql_query_summaries(
        getattr(request, "sql_query_summaries", None)
    ) or _plain_text_from_cached_askme(
        getattr(request, "text_analysis", None),
        getattr(request, "text_analysis_html", None),
    )
    if not summary:
        return (
            None,
            None,
            None,
            "Provide text_analysis or text_analysis_html from the ask response for cached export.",
        )

    parsed = service._parse_df_input(prior_input)
    if not parsed or len(parsed) == 0:
        return None, None, None

    table_df = pd.DataFrame(parsed)
    df_display = table_df.copy()
    df_display.index = df_display.index + 1

    fig = getattr(request, "chart", None)
    code = request.code
    mq = request.my_question or ""
    if fig is None and code and str(code).strip() not in ("", "N") and len(table_df) <= 100:
        try:
            cols = list(table_df.columns)
            fig = generate_chart_from_code(code, table_df, cols, mq, service)
        except Exception as e:
            print(f"[DOWNLOAD-EXCEL] cached @askme chart error: {e}")

    return summary, df_display, fig, None


async def _askme_excel_payload(
    service: Any,
    my_question: str,
    df_input: Any,
    code: Optional[str],
) -> Tuple[Optional[str], Optional[pd.DataFrame], Optional[Dict[str, Any]], Optional[str]]:
    """
    Same core behavior as ask-question @askme: ask_data_analysis_cached + prior df + optional chart from code.
    Returns (summary, df_display, fig, error_message).
    """
    if not getattr(settings, "IS_ASKME_ACTIVE", False):
        return None, None, None, "Askme analysis (@askme) is not enabled on this server."

    query = _strip_askme_token(my_question)
    if not query:
        return (
            None,
            None,
            None,
            "Please provide your analysis request after @askme (e.g. trend analysis, forecast, summary).",
        )

    parsed = service._parse_df_input(df_input)
    if not parsed or len(parsed) == 0:
        return (
            None,
            None,
            None,
            "No previous data found. Provide a dataframe in 'df' (same as ask-question) for @askme export.",
        )

    analysis_df = pd.DataFrame(parsed)

    analysis_result = await run_in_threadpool(
        service.ask_data_analysis_cached,
        question=query,
        df=analysis_df,
        use_qwen=settings.default_use_qwen_for_summary_and_web(),
    )

    analysis_text = None
    if analysis_result.get("text"):
        analysis_text = analysis_result["text"]
    elif not analysis_result.get("has_dataframe") and not analysis_result.get("has_chart"):
        analysis_text = str(analysis_result.get("text", "") or "")

    result_df = analysis_result.get("dataframe")
    if result_df is not None and hasattr(result_df, "empty") and not result_df.empty:
        table_df = result_df
    else:
        table_df = analysis_df

    _txt = (analysis_text or "").strip()
    _has_any = (
        analysis_result.get("has_text")
        or analysis_result.get("has_dataframe")
        or analysis_result.get("has_chart")
        or bool(_txt)
        or (analysis_result.get("dataframe") is not None)
    )
    if not _has_any:
        return (
            None,
            None,
            None,
            "Could not generate analysis. Please try rephrasing your request.",
        )

    df_display = None
    fig = None
    if table_df is not None and len(table_df) > 0:
        df_display = table_df.copy()
        df_display.index = df_display.index + 1
        if len(table_df) <= 100 and code and str(code).strip() not in ("", "N"):
            try:
                cols = list(table_df.columns)
                fig = generate_chart_from_code(code, table_df, cols, my_question, service)
            except Exception as e:
                print(f"[DOWNLOAD-EXCEL] @askme chart error: {e}")

    return analysis_text or _txt or None, df_display, fig, None


def generate_chart_image(chart_fig_dict: Dict[str, Any]) -> Optional[bytes]:
    """
    Generate a chart image from plotly figure dict using matplotlib (NO kaleido).
    Supports common Plotly trace types: bar, scatter(line), pie.
    """
    try:
        if not isinstance(chart_fig_dict, dict):
            return None

        data = chart_fig_dict.get("data", [])
        layout = chart_fig_dict.get("layout", {}) or {}

        title = ""
        if isinstance(layout.get("title"), dict):
            title = layout.get("title", {}).get("text", "") or ""
        else:
            title = layout.get("title", "") or ""

        xaxis_title = (layout.get("xaxis", {}) or {}).get("title", "")
        if isinstance(xaxis_title, dict):
            xaxis_title = xaxis_title.get("text", "") or ""

        yaxis_title = (layout.get("yaxis", {}) or {}).get("title", "")
        if isinstance(yaxis_title, dict):
            yaxis_title = yaxis_title.get("text", "") or ""

        # --- matplotlib setup ---
        plt.close("all")  # avoid memory leak in background jobs
        fig, ax = plt.subplots(figsize=(3, 2), dpi=200)

        default_colors = [
            "#e6194b", "#3cb44b", "#ffe119", "#4363d8", "#f58231",
            "#911eb4", "#46f0f0", "#f032e6", "#bcf60c", "#fabebe",
            "#008080", "#e6beff", "#9a6324", "#fffac8", "#800000",
            "#aaffc3", "#808000", "#ffd8b1", "#000075", "#808080",
        ]

        has_pie = False
        bar_traces = []  # Collect all bar traces for grouped rendering
        line_traces = []  # Collect line/scatter traces
        final_x_values = None
        
        # Helper: safely convert string numeric values to float
        def _to_numeric(vals):
            out = []
            for v in vals:
                if v is None:
                    out.append(0)
                elif isinstance(v, str):
                    try:
                        out.append(float(v))
                    except (ValueError, TypeError):
                        out.append(v)
                else:
                    out.append(v)
            return out

        # First pass: collect and categorize traces
        for trace in data:
            if not isinstance(trace, dict):
                continue
            
            trace_type = (trace.get("type") or "scatter").lower()
            
            if trace_type == "bar":
                x = trace.get("x", [])
                y = trace.get("y", [])
                x = list(x) if x is not None else []
                y = list(y) if y is not None else []
                
                # Convert None/string values to numeric
                y = _to_numeric(y)
                
                # Skip if all values are None/0
                if not any(y):
                    continue
                
                bar_traces.append({
                    "x": x,
                    "y": y,
                    "name": trace.get("name", f"Series {len(bar_traces)+1}"),
                    "color": trace.get("marker", {}).get("color") if isinstance(trace.get("marker"), dict) else None
                })
                if final_x_values is None:
                    final_x_values = x
                    
            elif trace_type == "pie":
                labels = trace.get("labels", [])
                values = trace.get("values", [])
                
                # Convert None/string values to numeric and filter
                values = _to_numeric(values)
                filtered_data = [(l, v) for l, v in zip(labels, values) if v and v > 0]
                if not filtered_data:
                    continue
                
                labels, values = zip(*filtered_data) if filtered_data else ([], [])
                ax.pie(values, labels=labels, autopct="%1.1f%%")
                ax.set_title(title or trace.get("name", ""))
                has_pie = True
                
            else:  # line/scatter
                mode = (trace.get("mode") or "lines").lower()
                x = trace.get("x", [])
                y = trace.get("y", [])
                x = list(x) if x is not None else []
                y = list(y) if y is not None else []
                
                # Convert None/string values to numeric
                y = _to_numeric(y)
                
                # Skip if all values are None/0
                if not any(y):
                    continue
                
                line_traces.append({
                    "x": x,
                    "y": y,
                    "name": trace.get("name", f"Series {len(line_traces)+1}"),
                    "mode": mode,
                    "color": trace.get("marker", {}).get("color") if isinstance(trace.get("marker"), dict) else None
                })
                if final_x_values is None:
                    final_x_values = x
        
        # Render grouped bar charts
        if bar_traces:
            num_bars = len(bar_traces)
            num_categories = len(bar_traces[0]["x"]) if bar_traces else 0
            
            # Calculate bar width and positions for grouped bars
            bar_width = 0.8 / num_bars if num_bars > 0 else 0.8
            x_positions = range(num_categories)
            
            for i, bar_trace in enumerate(bar_traces):
                # Calculate offset for this bar series
                offset = (i - num_bars / 2) * bar_width + bar_width / 2
                positions = [x + offset for x in x_positions]
                
                # Get color
                color = bar_trace["color"]
                if not isinstance(color, str) or not color.startswith("#"):
                    color = default_colors[i % len(default_colors)]
                
                ax.bar(positions, bar_trace["y"], width=bar_width, 
                       label=bar_trace["name"], color=color)
        
        # Render line/scatter charts
        for i, line_trace in enumerate(line_traces):
            idx = range(len(line_trace["x"]))
            
            # Get color
            color = line_trace["color"]
            if not isinstance(color, str) or not color.startswith("#"):
                color = default_colors[(len(bar_traces) + i) % len(default_colors)]
            
            mode = line_trace["mode"]
            if "markers" in mode and "lines" in mode:
                ax.plot(idx, line_trace["y"], marker="o", label=line_trace["name"], color=color)
            elif "markers" in mode:
                ax.scatter(idx, line_trace["y"], label=line_trace["name"], color=color)
            else:
                ax.plot(idx, line_trace["y"], label=line_trace["name"], color=color)
        
        series_count = len(bar_traces) + len(line_traces)

        if not has_pie:
            if title:
                ax.set_title(title, fontsize=8)
            if xaxis_title:
                ax.set_xlabel(xaxis_title, fontsize=6)
            if yaxis_title:
                ax.set_ylabel(yaxis_title, fontsize=6)

            # Set x-axis tick labels to show actual data labels
            if final_x_values:
                # Set tick positions at the center of each group
                ax.set_xticks(range(len(final_x_values)))
                # Set tick labels to the actual x values from data
                ax.set_xticklabels(final_x_values, rotation=45, ha='right', fontsize=5)
            
            # Set y-axis tick label font size
            ax.tick_params(axis='y', labelsize=5)
            # Disable scientific notation on y-axis - show actual values
            try:
                ax.ticklabel_format(style='plain', axis='y')
            except (AttributeError, ValueError):
                pass  # axis not using ScalarFormatter (e.g. string/categorical data)

            # show legend only if multiple series
            if series_count > 1:
                ax.legend(loc="best", fontsize=5)

            ax.grid(True, linestyle="--", linewidth=0.5, alpha=0.4)

        fig.tight_layout()

        buf = io.BytesIO()
        fig.savefig(buf, format="png", bbox_inches="tight")
        plt.close(fig)
        buf.seek(0)
        return buf.read()

    except Exception as e:
        print(f"Error generating chart image (matplotlib): {e}")
        return None

# import pandas as pd
# import numpy as np

# def format_x_labels(ax, x_values, max_labels=8, rotate=0):
#     labels = [str(v) for v in (x_values or [])]

#     # Try parsing as datetime
#     try:
#         dt = pd.to_datetime(labels, errors="coerce")
#         dt = pd.Series(dt)   # ✅ FIX: ensures .dt works

#         if dt.notna().sum() >= len(labels) * 0.6 and len(labels) > 0:
#             years = dt.dt.year.dropna().unique()

#             if len(years) == 1:
#                 labels = dt.dt.strftime("%b").tolist()        # Jan Feb...
#             else:
#                 labels = dt.dt.strftime("%b %Y").tolist()     # Jan 2024...
#     except Exception:
#         pass

#     n = len(labels)
#     if n == 0:
#         return

#     step = max(1, n // max_labels)
#     tick_positions = np.arange(0, n, step)

#     ax.set_xticks(tick_positions)
#     ax.set_xticklabels(
#         [labels[i] for i in tick_positions],
#         rotation=rotate,
#         ha="center"
#     )

def generate_chart_from_code(code: str, df: pd.DataFrame, columns: List[str], question: str, service) -> Optional[Dict[str, Any]]:
    """Generate a plotly figure from Python code string."""
    try:
        import json as json_module
        # Execute the code in a safe namespace
        # Make json_module available in execution context
        exec_globals = {
            "pd": pd,
            "json": json_module,
            "json_module": json_module,
            "go": go,
            "px": px,
        }
        local_vars = {
            "df": df,
            "pd": pd,
            "json": json_module,
            "json_module": json_module,
            "go": go,
            "px": px,
        }
        exec(code, exec_globals, local_vars)
        
        # Extract the chart_json variable created by the code
        chart_json = local_vars.get("chart_json")
        # print("changing chart_json", chart_json)
        return chart_json
    except Exception as e:
        print(f"[DOWNLOAD-EXCEL] Error generating chart from code: {e}")
        # Return default chart on error
        return None


def create_excel_file(
    question: str,
    chart_fig: Optional[Dict[str, Any]] = None,
    data: Optional[Union[pd.DataFrame, List[pd.DataFrame]]] = None,
    summary: Optional[str] = None,
    agent_names: Optional[List[str]] = None
) -> io.BytesIO:
    """Create an Excel file with text, charts, and dataframes."""
    # Create a new workbook
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "askme360 Response"
    
    # Set Excel title
    ws['A1'] = "askme360 - Chat Response"
    ws['A1'].font = Font(size=16, bold=True)
    ws['A2'] = f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    
    # Add the original question
    ws['A4'] = "Question:"
    ws['A4'].font = Font(bold=True)
    ws['A5'] = question
    
    current_row = 7  # Start after the question
    
    # Add chart if provided
    if chart_fig:
        ws.cell(row=current_row, column=1).value = "Visualization:"
        ws.cell(row=current_row, column=1).font = Font(bold=True)
        current_row += 1
        
        try:
            # Generate chart image
            img_bytes = generate_chart_image(chart_fig)
            
            if img_bytes:
                # Use BytesIO instead of temporary file to avoid file deletion issues
                img_buffer = io.BytesIO(img_bytes)
                
                # Insert image into Excel from BytesIO
                img = Image(img_buffer)
                
                # Resize image if too large (Excel has limitations)
                if img.width > 800:
                    ratio = img.height / img.width
                    img.width = 800
                    img.height = int(800 * ratio)
                
                ws.add_image(img, f'A{current_row}')
                
                # Move row pointer past the image
                # Approximate number of rows the image will take
                image_rows = max(1, int(img.height / 20))  # 20 pixels per row (approximate)
                current_row += image_rows + 2
        except Exception as e:
            ws.cell(row=current_row, column=1).value = "Failed to insert chart image."
            ws.cell(row=current_row + 1, column=1).value = str(e)
            current_row += 3
    
    # Add data if provided
    # Keep SQL summary above tabular output for readability.
    if summary:
        _summary_heading = (
            "SQL Summary:"
            if ("Sub Question:" in str(summary) and "Summary:" in str(summary))
            else "Response:"
        )
        ws.cell(row=current_row, column=1).value = _summary_heading
        ws.cell(row=current_row, column=1).font = Font(bold=True)
        current_row += 1

        text_lines = summary.split('\n')
        for line in text_lines:
            ws.cell(row=current_row, column=1).value = line
            current_row += 1
        current_row += 1  # Add extra space

    # Add data if provided
    if data is not None:
        # Convert to list to uniformly process single or multiple dataframes
        dfs = data if isinstance(data, list) else [data]
        
        for idx, df in enumerate(dfs):
            if df.empty:
                continue

            if agent_names and idx < len(agent_names) and agent_names[idx]:
                ws.cell(row=current_row, column=1).value = f"Data Results - {str(agent_names[idx]).strip()}:"
            else:
                ws.cell(row=current_row, column=1).value = f"Data Results{' - View ' + str(idx + 1) if len(dfs) > 1 else ''}:"
            ws.cell(row=current_row, column=1).font = Font(bold=True)
            current_row += 1
            
            # Only include first 1000 rows and first 1000 columns for Excel file
            display_df = df.iloc[:1000, :1000] if df.shape[1] > 1000 else df.iloc[:1000]
            
            # Create a table from dataframe
            start_row = current_row
            
            # Add headers with formatting
            header_fill = PatternFill(start_color="DDDDDD", end_color="DDDDDD", fill_type="solid")
            
            for c, col in enumerate(display_df.columns, 1):
                cell = ws.cell(row=current_row, column=c)
                cell.value = str(col)
                cell.font = Font(bold=True)
                cell.fill = header_fill
                cell.alignment = Alignment(wrap_text=True)
            current_row += 1
            
            # Add data rows
            for _, row_data in display_df.iterrows():
                for c, value in enumerate(row_data, 1):
                    cell = ws.cell(row=current_row, column=c)
                    cell.value = str(value) if not pd.isna(value) else ""
                    cell.alignment = Alignment(wrap_text=True)
                current_row += 1
            
            # Auto-adjust column widths based on content
            for c in range(1, len(display_df.columns) + 1):
                max_length = 0
                column = get_column_letter(c)
                
                # Check header length
                header_text = str(display_df.columns[c-1])
                if len(header_text) > max_length:
                    max_length = len(header_text)
                
                # Check data length in each row (up to 50 rows to safely guess width)
                for row_num in range(start_row + 1, min(current_row, start_row + 50)):
                    cell_value = ws.cell(row=row_num, column=c).value
                    if cell_value:
                        cell_text = str(cell_value)
                        if len(cell_text) > max_length:
                            max_length = min(len(cell_text), 50)  # Cap width at 50 characters
                
                # Only increase width if it's larger than the current, to not shrink earlier table columns
                current_width = ws.column_dimensions[column].width or 0
                adjusted_width = max_length + 2
                if adjusted_width > current_width:
                    ws.column_dimensions[column].width = adjusted_width
            
            # Add note about truncation if needed
            if len(df) > 1000:
                current_row += 1
                ws.cell(row=current_row, column=1).value = f"Note: Showing first 1000 rows of {len(df)} total rows"
                ws.cell(row=current_row, column=1).font = Font(italic=True)
            
            current_row += 2  # Add extra space after table
    
    # Save Excel file to in-memory buffer
    excel_buffer = io.BytesIO()
    wb.save(excel_buffer)
    excel_buffer.seek(0)
    
    return excel_buffer


def _collect_non_empty_frames(
    value: Any, order: Optional[List[str]] = None
) -> List[pd.DataFrame]:
    """Normalize DataFrame/list/dict payloads into a list of non-empty DataFrames."""
    frames: List[pd.DataFrame] = []
    if value is None:
        return frames
    if isinstance(value, pd.DataFrame):
        if not value.empty:
            frames.append(value)
        return frames
    if isinstance(value, list):
        for item in value:
            if isinstance(item, pd.DataFrame) and not item.empty:
                frames.append(item)
        return frames
    if isinstance(value, dict):
        keys: List[Any] = []
        if order:
            keys.extend([k for k in order if k in value])
        keys.extend([k for k in value.keys() if k not in keys])
        for k in keys:
            item = value.get(k)
            if isinstance(item, pd.DataFrame) and not item.empty:
                frames.append(item)
        return frames
    return frames


def _write_sectioned_frames_to_sheet(
    ws: Any,
    source_name: str,
    sections: List[Tuple[str, pd.DataFrame]],
    start_row: int = 1,
) -> int:
    """Write stacked dataframe sections (title + headers + rows) into one sheet."""
    current_row = start_row
    header_fill = PatternFill(start_color="DDDDDD", end_color="DDDDDD", fill_type="solid")

    for idx, (label, src_df) in enumerate(sections, start=1):
        if src_df is None or getattr(src_df, "empty", True):
            continue

        safe_label = (str(label).strip() or f"{source_name} View {idx}")
        ws.cell(row=current_row, column=1).value = f"Data Results - {safe_label}"
        ws.cell(row=current_row, column=1).font = Font(bold=True)
        current_row += 1

        ws.cell(row=current_row, column=1).value = (
            f"Source: {source_name} | Rows: {len(src_df)}"
        )
        ws.cell(row=current_row, column=1).font = Font(italic=True)
        current_row += 1

        display_df = (
            src_df.iloc[:1000, :1000] if src_df.shape[1] > 1000 else src_df.iloc[:1000]
        )
        start_table_row = current_row

        for c, col in enumerate(display_df.columns, 1):
            cell = ws.cell(row=current_row, column=c)
            cell.value = str(col)
            cell.font = Font(bold=True)
            cell.fill = header_fill
            cell.alignment = Alignment(wrap_text=True)
        current_row += 1

        for _, row_data in display_df.iterrows():
            for c, value in enumerate(row_data, 1):
                cell = ws.cell(row=current_row, column=c)
                cell.value = str(value) if not pd.isna(value) else ""
                cell.alignment = Alignment(wrap_text=True)
            current_row += 1

        for c in range(1, len(display_df.columns) + 1):
            max_length = len(str(display_df.columns[c - 1]))
            col_letter = get_column_letter(c)
            for row_num in range(start_table_row + 1, min(current_row, start_table_row + 50)):
                cell_value = ws.cell(row=row_num, column=c).value
                if cell_value:
                    max_length = max(max_length, min(len(str(cell_value)), 50))
            current_width = ws.column_dimensions[col_letter].width or 0
            adjusted_width = max_length + 2
            if adjusted_width > current_width:
                ws.column_dimensions[col_letter].width = adjusted_width

        if len(src_df) > 1000:
            ws.cell(row=current_row, column=1).value = (
                f"Note: Showing first 1000 rows of {len(src_df)} total rows"
            )
            ws.cell(row=current_row, column=1).font = Font(italic=True)
            current_row += 1

        current_row += 2

    return current_row


def create_excel_file_by_source(
    question: str,
    db_sections: List[Tuple[str, pd.DataFrame]],
    api_sections: List[Tuple[str, pd.DataFrame]],
    chart_fig: Optional[Dict[str, Any]] = None,
    summary: Optional[str] = None,
) -> io.BytesIO:
    """
    Create workbook with separate source sheets and stacked sections:
      - DATABASE sheet for DB frames
      - API sheet for API frames
    """
    wb = openpyxl.Workbook()
    default_ws = wb.active
    default_ws.title = "Summary"

    ws_summary = wb["Summary"]
    ws_summary["A1"] = "askme360 - Chat Response"
    ws_summary["A1"].font = Font(size=16, bold=True)
    ws_summary["A2"] = f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    ws_summary["A4"] = "Question:"
    ws_summary["A4"].font = Font(bold=True)
    ws_summary["A5"] = question
    s_row = 7

    if summary:
        _summary_heading = (
            "SQL Summary:"
            if ("Sub Question:" in str(summary) and "Summary:" in str(summary))
            else "Response:"
        )
        ws_summary.cell(row=s_row, column=1).value = _summary_heading
        ws_summary.cell(row=s_row, column=1).font = Font(bold=True)
        s_row += 1
        for line in str(summary).split("\n"):
            ws_summary.cell(row=s_row, column=1).value = line
            s_row += 1
        s_row += 1

    if chart_fig:
        ws_summary.cell(row=s_row, column=1).value = "Visualization:"
        ws_summary.cell(row=s_row, column=1).font = Font(bold=True)
        s_row += 1
        try:
            img_bytes = generate_chart_image(chart_fig)
            if img_bytes:
                img = Image(io.BytesIO(img_bytes))
                if img.width > 800:
                    ratio = img.height / img.width
                    img.width = 800
                    img.height = int(800 * ratio)
                ws_summary.add_image(img, f"A{s_row}")
        except Exception as e:
            ws_summary.cell(row=s_row, column=1).value = f"Failed to insert chart image: {e}"

    if db_sections:
        ws_db = wb.create_sheet(title="DATABASE")
        _write_sectioned_frames_to_sheet(ws_db, "DATABASE", db_sections, start_row=1)
    if api_sections:
        ws_api = wb.create_sheet(title="API")
        _write_sectioned_frames_to_sheet(ws_api, "API", api_sections, start_row=1)

    excel_buffer = io.BytesIO()
    wb.save(excel_buffer)
    excel_buffer.seek(0)
    return excel_buffer


def get_mail_content(schedule_type=None, activity=None, description=None):
    """
    Generate email content using predefined templates for scheduled or regular responses.
    
    Args:
        schedule_type (str, optional): Type of schedule (e.g., "Daily", "Weekly", "Monthly"). 
                                     If provided, uses scheduled template
        activity (str): Activity description (question)
        description (str, optional): Additional description text
    
    Returns:
        tuple: (body_html, subject)
    """
    
    # Get current date and time
    current_time = datetime.now()
    date_str = current_time.strftime("%B %d, %Y")
    time_str = current_time.strftime("%I:%M %p IST")
    
    # Determine which template to use based on schedule_type
    if schedule_type is not None:
        # Scheduled email template
        subject = f"{schedule_type} Scheduled Response from askme360"
        
        body_html = f"""
        <div style="font-family: Poppins, sans-serif; margin: 0 auto;">
            <p>Dear user,</p>
            
            <p>Please find attached the requested <strong>{schedule_type}</strong> report.</p>
            
            </br>
            <p><strong>Schedule Type:</strong> {schedule_type}</p>
            <p><strong>Date:</strong> {date_str}</p>
            <p><strong>Time Generated:</strong> {time_str}</p>
        
            </br>
            <p><strong>Activity:</strong> {activity}</p>
            
            {f'''<div style="margin: 20px 0;">
                <p><strong>Description Text:</strong></p>
                <p>{description.strip()}</p>
            </div>''' if description and description.strip() else ''}
            
            <div style="background-color: #fffbcc; padding: 10px; margin: 20px 0; border-left: 4px solid #ffd700;">
                <p><em>"This is an automated message. Please do not reply."</em></p>
            </div>
            
            <p>Thank you<br>
            <strong>askme360</strong></p>
        </div>
        """
    
        
    else:
        # Regular response template
        subject = "Response from askme360"
        
        body_html = f"""
        <div style="font-family: Arial, sans-serif; margin: 0 auto;">
            <p>Dear user,</p>
            
            <p>Please find attached the requested report.</p>
            </br>
            <p><strong>Date:</strong> {date_str}</p>
            <p><strong>Time Generated:</strong> {time_str}</p>
             </br>
            <p><strong>Activity:</strong> {activity}</p>
            
            {f'''<div style="margin: 20px 0;">
                <p><strong>Description Text:</strong></p>
                <p>{description}</p>
            </div>''' if description is not None else ''}
            
            <div style="background-color: #fffbcc; padding: 10px; margin: 20px 0; border-left: 4px solid #ffd700;">
                <p><em>"This is an automated message. Please do not reply."</em></p>
            </div>
            
            <p>Thank you<br>
            <strong>askme360</strong></p>
        </div>
        """
     
    return body_html, subject


_ALLOWED_IMAGE_PREFIXES = ("document_images_", "extracted_images_", "uploaded_files_")


def _file_id_page_from_chunk_id(chunk_id: Optional[str]) -> Tuple[Optional[str], Optional[int]]:
    if not chunk_id:
        return None, None
    cid = str(chunk_id).strip()
    if "_page_" not in cid:
        return None, None
    base, _, page_s = cid.rpartition("_page_")
    if not base or not page_s.isdigit():
        return None, None
    return base, int(page_s)


def _parse_page_media_api_url(path: str) -> Tuple[Optional[str], Optional[str], Optional[int]]:
    """Parse runtime page-image / download-page API URLs."""
    if not path:
        return None, None, None
    s = str(path).strip()
    for marker in ("/document/page-image/", "/document/download-page/"):
        if marker not in s:
            continue
        tail = s.split(marker, 1)[1]
        file_id = unquote(tail.split("?", 1)[0].strip().strip("/"))
        qs = parse_qs(urlparse(s).query)
        agent_id = (qs.get("agent_id") or [None])[0]
        page_raw = (qs.get("page_number") or [None])[0]
        page_n = int(page_raw) if page_raw and str(page_raw).isdigit() else None
        return file_id or None, str(agent_id).strip() if agent_id else None, page_n
    return None, None, None


def _enrich_document_email_image_rows(
    images: Optional[List[Dict[str, Any]]],
    *,
    agent_id: Optional[str] = None,
    chunk_id: Optional[str] = None,
    file_id: Optional[str] = None,
) -> List[Dict[str, Any]]:
    chunk_fid, chunk_page = _file_id_page_from_chunk_id(chunk_id)
    default_fid = file_id or chunk_fid
    default_aid = str(agent_id).strip() if agent_id else None
    out: List[Dict[str, Any]] = []
    for img in images or []:
        if not isinstance(img, dict):
            continue
        row = dict(img)
        parsed_fid, parsed_aid, parsed_page = _parse_page_media_api_url(row.get("path") or "")
        if not row.get("file_id"):
            row["file_id"] = parsed_fid or default_fid
        if not row.get("agent_id"):
            row["agent_id"] = parsed_aid or default_aid
        if row.get("page") is None:
            row["page"] = parsed_page if parsed_page is not None else chunk_page
        out.append(row)
    return out


def _resolve_disk_image_path(path: str) -> Optional[str]:
    """Resolve a URL or relative image path to an on-disk file under cwd."""
    if not path:
        return None
    path = str(path).strip().strip('"').strip("'")
    for marker in ("/api/images/", "/api/static/images/", "/static/images/"):
        if marker in path:
            path = path.split(marker, 1)[-1]
    path = path.replace("\\", "/")
    rel = None
    for prefix in _ALLOWED_IMAGE_PREFIXES:
        if path.startswith(prefix):
            rel = path
            break
        idx = path.find(prefix)
        if idx != -1:
            rel = path[idx:]
            break
    if not rel:
        return None
    full_path = os.path.join(os.getcwd(), rel.replace("/", os.sep))
    if os.path.isfile(full_path):
        return full_path
    return None


def _render_page_png_disk(
    agent_id: str,
    file_id: str,
    page_number: int,
) -> Optional[str]:
    """Render a single PDF page to a temporary PNG for email attachments."""
    try:
        from app.services.document_db_service import DocumentDbService
        from app.utils.document_preview import (
            DEFAULT_PREVIEW_PDF_CACHE_DIR,
            render_page_png_bytes,
            resolve_preview_file_response,
        )

        db_svc = DocumentDbService(db_file="file_sys_metadata.db", agent_id=agent_id)
        try:
            row = db_svc.get_file(file_id, agent_id=agent_id)
            if not row or len(row) < 5:
                return None
            file_path = row[4]
            filename = row[2] if len(row) > 2 else os.path.basename(file_path)
            full_pdf_path, media_type, _ = resolve_preview_file_response(
                file_path,
                filename,
                file_id,
                inline=True,
                cache_dir=DEFAULT_PREVIEW_PDF_CACHE_DIR,
            )
            if media_type != "application/pdf":
                return None
            png_bytes = render_page_png_bytes(full_pdf_path, int(page_number))
        finally:
            db_svc.close()
        fd, tmp_path = tempfile.mkstemp(suffix=".png", prefix="doc_email_page_")
        os.close(fd)
        with open(tmp_path, "wb") as tmp_file:
            tmp_file.write(png_bytes)
        return tmp_path
    except Exception as exc:
        print(f"[SEND-DOCUMENT-EMAIL] page PNG render error: {exc}")
        return None


def _render_page_pdf_disk(
    agent_id: str,
    file_id: str,
    page_number: int,
) -> Optional[str]:
    """Build cached single-page PDF on disk for email attachment."""
    try:
        from app.services.document_db_service import DocumentDbService
        from app.utils.document_preview import render_page_pdf_for_download

        db_svc = DocumentDbService(db_file="file_sys_metadata.db", agent_id=agent_id)
        try:
            row = db_svc.get_file(file_id, agent_id=agent_id)
            if not row or len(row) < 5:
                return None
            file_path = row[4]
            filename = row[2] if len(row) > 2 else os.path.basename(file_path)
            return render_page_pdf_for_download(
                file_path,
                filename,
                file_id,
                page_number,
            )
        finally:
            db_svc.close()
    except Exception as exc:
        print(f"[SEND-DOCUMENT-EMAIL] page PDF render error: {exc}")
        return None


def _collect_document_image_disk_paths(
    images: Optional[List[Dict[str, Any]]],
    *,
    agent_id: Optional[str] = None,
    chunk_id: Optional[str] = None,
    file_id: Optional[str] = None,
) -> List[str]:
    """Collect unique on-disk paths (legacy images or runtime-rendered page PNGs)."""
    items = _enrich_document_email_image_rows(
        [img for img in (images or []) if isinstance(img, dict)],
        agent_id=agent_id,
        chunk_id=chunk_id,
        file_id=file_id,
    )
    items.sort(key=lambda x: (x.get("page") is None, x.get("page") or 0))

    paths: List[str] = []
    seen = set()
    for img in items:
        disk = None
        fid = img.get("file_id")
        aid = img.get("agent_id")
        page = img.get("page")
        if fid and aid and page is not None:
            disk = _render_page_png_disk(str(aid), str(fid), int(page))
        if not disk and img.get("path"):
            disk = _resolve_disk_image_path(str(img.get("path")))
        if disk and disk not in seen:
            seen.add(disk)
            paths.append(disk)
    return paths


def _build_pdf_from_page_images(
    page_images: Any,
    attachment_id: str,
    agent_id: Optional[str] = None,
    file_id: Optional[str] = None,
) -> Tuple[Optional[str], List[str]]:
    """@find: convert page_images dict to a single PDF attachment."""
    normalized = _normalize_document_email_images(page_images=page_images)
    return _build_document_email_pdf(
        normalized,
        attachment_id,
        agent_id=agent_id,
        file_id=file_id,
    )


def _build_pdf_from_image_paths(image_paths: List[str], output_path: str) -> bool:
    """Combine image files into a single PDF attachment."""
    if not image_paths:
        return False
    from PIL import Image

    pil_images = []
    try:
        for path in image_paths:
            with Image.open(path) as img:
                pil_images.append(img.convert("RGB"))
        if len(pil_images) == 1:
            pil_images[0].save(output_path, format="PDF")
        else:
            pil_images[0].save(
                output_path,
                format="PDF",
                save_all=True,
                append_images=pil_images[1:],
            )
        return os.path.isfile(output_path)
    except Exception as e:
        print(f"[SEND-DOCUMENT-CHUNK-EMAIL] PDF build error: {e}")
        return False
    finally:
        for img in pil_images:
            img.close()


def _build_document_email_pdf(
    images: Optional[List[Dict[str, Any]]],
    attachment_id: str,
    agent_id: Optional[str] = None,
    file_id: Optional[str] = None,
) -> Tuple[Optional[str], List[str]]:
    """Build a PDF from normalized document images; returns (pdf_path, source_image_paths)."""
    image_paths = _collect_document_image_disk_paths(
        images,
        agent_id=agent_id,
        chunk_id=attachment_id,
        file_id=file_id,
    )
    if not image_paths:
        return None, []

    email_dir = os.path.join(os.getcwd(), "email_files")
    os.makedirs(email_dir, exist_ok=True)
    safe_id = re.sub(r"[^\w\-]+", "_", attachment_id or "document")
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    pdf_filename = f"document_{safe_id}_{timestamp}.pdf"
    pdf_path = os.path.join(email_dir, pdf_filename)

    if not _build_pdf_from_image_paths(image_paths, pdf_path):
        return None, image_paths
    return pdf_path, image_paths


_NO_IMAGES_PRESENT_MSG = "no document pages present"


async def _process_document_email(
    *,
    agent_id: str,
    sendemail: str,
    email_list: Optional[List[str]],
    my_question: Optional[str],
    document_result_type: Optional[str] = None,
    relevant_images: Optional[List[Dict[str, Any]]] = None,
    page_images: Optional[Dict[str, List[Dict[str, Any]]]] = None,
    document_search: Optional[Dict[str, Any]] = None,
    document_find: Optional[Dict[str, Any]] = None,
    chunk_id: Optional[str] = None,
    file_id: Optional[str] = None,
    filename: Optional[str] = None,
    agent_config: dict,
    service,
) -> dict:
    """Build a PDF from @file or @find images and send email; never attach Excel."""
    (
        document_result_type,
        relevant_images,
        page_images,
        chunk_id,
        file_id,
        filename,
    ) = await _hydrate_document_email_images_from_db(
        agent_id=agent_id,
        my_question=my_question,
        document_result_type=document_result_type,
        relevant_images=relevant_images,
        page_images=page_images,
        document_search=document_search,
        document_find=document_find,
        chunk_id=chunk_id,
        file_id=file_id,
        filename=filename,
        agent_config=agent_config,
    )

    images, resolved_filename, resolved_chunk_id, resolved_file_id = _resolve_document_email_payload(
        document_result_type=document_result_type,
        relevant_images=relevant_images,
        page_images=page_images,
        document_search=document_search,
        document_find=document_find,
        chunk_id=chunk_id,
        file_id=file_id,
        filename=filename,
    )
    attachment_id = resolved_chunk_id or resolved_file_id or chunk_id or file_id or ""
    source_image_paths = _collect_document_image_disk_paths(
        images,
        agent_id=agent_id,
        chunk_id=attachment_id,
        file_id=resolved_file_id,
    )

    if sendemail != "Y":
        return {
            "success": True,
            "message": "Email not requested",
            "details": {
                "chunk_id": resolved_chunk_id,
                "file_id": resolved_file_id,
                "document_result_type": document_result_type,
                "email_sent": False,
                "relevant_images_found": len(source_image_paths),
            },
        }

    if not email_list:
        return {
            "success": False,
            "message": "email_list is required when sendemail is Y",
            "details": {
                "chunk_id": resolved_chunk_id,
                "file_id": resolved_file_id,
                "email_sent": False,
            },
        }

    if not source_image_paths:
        return {
            "success": False,
            "message": _NO_IMAGES_PRESENT_MSG,
            "details": {
                "chunk_id": resolved_chunk_id,
                "file_id": resolved_file_id,
                "document_result_type": document_result_type,
                "email_sent": False,
                "relevant_images_found": 0,
            },
        }

    find_pages = _coerce_page_images(page_images)
    if document_result_type == "document_find" and find_pages:
        pdf_path, source_image_paths = await run_in_threadpool(
            _build_pdf_from_page_images,
            find_pages,
            attachment_id,
            agent_id,
            resolved_file_id,
        )
    else:
        pdf_path, source_image_paths = await run_in_threadpool(
            _build_document_email_pdf,
            images,
            attachment_id,
            agent_id,
            resolved_file_id,
        )
    if not pdf_path:
        return {
            "success": False,
            "message": _NO_IMAGES_PRESENT_MSG,
            "details": {
                "chunk_id": resolved_chunk_id,
                "file_id": resolved_file_id,
                "document_result_type": document_result_type,
                "email_sent": False,
                "relevant_images_found": len(source_image_paths),
            },
        }

    activity = my_question or resolved_filename or attachment_id
    body_html, subject = get_mail_content(
        schedule_type=None,
        activity=activity,
        description="Please find the attached images",
    )
    if resolved_filename:
        subject = f"Document images: {resolved_filename} ({attachment_id})"

    email_sent = False
    email_error = None
    try:
        email_sent = await run_in_threadpool(
            service.send_email,
            sender_email=agent_config.get("sender_email"),
            sender_name="askme360",
            recipient_email=email_list,
            subject=subject,
            body_text="Please find the attached images",
            body_html=body_html,
            attachment_path=pdf_path,
            agent_conf=agent_config,
        )
        if not email_sent:
            email_error = "Failed to send email. Please check SMTP configuration."
    except Exception as e:
        email_error = f"Error sending email: {str(e)}"
        print(f"[SEND-DOCUMENT-EMAIL] Email error: {email_error}")

    details = {
        "chunk_id": resolved_chunk_id,
        "file_id": resolved_file_id,
        "document_result_type": document_result_type,
        "email_sent": email_sent,
        "relevant_images_included": len(source_image_paths),
        "attachment_name": os.path.basename(pdf_path),
        "attachment_type": "application/pdf",
    }
    if email_sent:
        details["email_recipients"] = email_list
    if email_error:
        details["email_error"] = email_error

    return {
        "success": email_sent,
        "message": (
            "Document PDF emailed successfully"
            if email_sent
            else (email_error or "Failed to send document email")
        ),
        "details": details,
    }


async def _process_document_email_from_download_request(
    request: DownloadExcelRequest,
    agent_id: str,
    agent_config: dict,
    service,
) -> dict:
    """Route download-excel document email using unified @file / @find payload fields."""
    return await _process_document_email(
        agent_id=agent_id,
        sendemail=request.sendemail,
        email_list=request.email_list,
        my_question=_resolve_download_question_text(request),
        document_result_type=getattr(request, "document_result_type", None),
        relevant_images=getattr(request, "relevant_images", None),
        page_images=getattr(request, "page_images", None),
        document_search=getattr(request, "document_search", None),
        document_find=getattr(request, "document_find", None),
        chunk_id=getattr(request, "chunk_id", None),
        file_id=getattr(request, "file_id", None),
        filename=getattr(request, "filename", None),
        agent_config=agent_config,
        service=service,
    )


@router.post("/send-document-chunk-email")
async def send_document_chunk_email(
    request: DocumentEmailRequest,
    current_user: User = Depends(get_current_user),
):
    """
    Send an email with a PDF attachment for @file or @find document results.

    @file: pass ``chunk_id`` + ``relevant_images`` (or ``document_search`` block).
    @find: pass ``file_id`` + ``page_images`` (or ``document_find`` block).
    """
    try:
        agent_id = str(request.agent_id or "").strip()
        if not agent_id:
            return {
                "success": False,
                "message": "agent_id is required",
                "details": {},
            }

        agent_config = get_agent_config(agent_id=agent_id)
        if not agent_config:
            return {
                "success": False,
                "message": f"Agent configuration not found for agent_id: {agent_id}",
                "details": {},
            }

        service = get_askme_service(agent_id=agent_id)
        if not service:
            return {
                "success": False,
                "message": f"Failed to initialize service for agent '{agent_id}'",
                "details": {},
            }

        return await _process_document_email(
            agent_id=agent_id,
            sendemail=request.sendemail,
            email_list=request.email_list,
            my_question=request.my_question or request.question,
            document_result_type=request.document_result_type,
            relevant_images=request.relevant_images,
            page_images=request.page_images,
            document_search=request.document_search,
            document_find=request.document_find,
            chunk_id=request.chunk_id,
            file_id=request.file_id,
            filename=request.filename,
            agent_config=agent_config,
            service=service,
        )

    except Exception as e:
        return {
            "success": False,
            "message": f"Error sending document email: {str(e)}",
            "details": {
                "chunk_id": getattr(request, "chunk_id", None),
                "file_id": getattr(request, "file_id", None),
            },
        }


@router.post("/download-excel")
async def download_excel(
    request: DownloadExcelRequest,
    http_request: Request,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """
    Download Excel file with question response data.
    
    Body:
        my_question: The question text (required)
        agent_id: Agent ID (required)
        sendemail: Send email flag - "Y" or "N" (required)
        email_list: List of email addresses (optional)
        sql: SQL query (optional)
        code: Chart code (optional)
        api_endpoint: API endpoint (optional)
        api_params: API parameters (optional)
        api_response_validator: API response validator (optional)
        df: Prior table for ``@askme`` (required for that path) and optional for ``@web`` (same shapes as ask-question)
    
    When no saved API/SQL is used, export uses the same pipelines as ask-question for ``@web`` and ``@askme`` only.
    
    Returns:
        Excel file download response
    """

    try:
        # If question_id is provided, fetch saved question and reuse SQL/API (same as ask-question)
        if request.question_id:
            try:
                user_id = str(getattr(current_user, "id", "")) or getattr(current_user, "user_id", "")
                question = db.query(Question).filter(
                    Question.id == request.question_id,
                    Question.user_id == user_id,
                ).first()
                if question:
                    # Populate request from saved question so we reuse SQL/API without regenerating
                    request.my_question = question.question or request.my_question
                    request.agent_id = question.agent_id or request.agent_id
                    request.sql = question.sql or request.sql
                    request.code = question.chart or request.code
                    request.api_endpoint = question.api_endpoint or request.api_endpoint
                    request.api_params = question.api_params or request.api_params
                    request.api_response_validator = question.api_response_validator or request.api_response_validator
                    request.sql_query_summaries = (
                        getattr(question, "sql_query_summaries", None)
                        or request.sql_query_summaries
                    )
                else:
                    return {
                        "success": False,
                        "message": f"Question not found for question_id: {request.question_id}",
                        "details": {}
                    }
            except Exception as e:
                print(f"Error fetching question data: {e}")
                return {
                    "success": False,
                    "message": f"Error fetching question data: {str(e)}",
                    "details": {}
                }
        else:
            try:
                user_id = str(getattr(current_user, "id", "")) or getattr(current_user, "user_id", "")
                await run_in_threadpool(_hydrate_request_from_user_question_log_cache, request, user_id)
            except Exception:
                pass
        
        # Resolve agent_id: for the single-agent endpoint always use the first (or only) agent
        if isinstance(request.agent_id, list):
            single_agent_id = request.agent_id[0] if request.agent_id else ""
        else:
            single_agent_id = request.agent_id

        # Get agent configuration

        agent_config = get_agent_config(agent_id=single_agent_id)
        if not agent_config:
            return {
                "success": False,
                "message": f"Agent configuration not found for agent_id: {single_agent_id}",
                "details": {}
            }
      
        # Get AskmeService instance

        service = get_askme_service(agent_id=single_agent_id)
        if not service:
            return {
                "success": False,
                "message": f"Failed to initialize service for agent '{single_agent_id}'",
                "details": {}
            }

        if _is_document_question(
            _resolve_download_question_text(request),
            getattr(request, "document_result_type", None),
            getattr(request, "is_file", None),
        ):
            if request.sendemail == "Y" and request.email_list:
                return await _process_document_email_from_download_request(
                    request,
                    single_agent_id,
                    agent_config,
                    service,
                )
            return {
                "success": False,
                "message": "Excel export is not supported for document questions.",
                "details": {},
            }

        
        df = None
        fig = None
        df_display = None
        summary = None
        
        
        # Parse API fields from request (reuse when provided in body or from question_id)
        api_endpoints = parse_api_field(request.api_endpoint)
        api_params_list = parse_api_field(request.api_params)
        api_validators = parse_api_field(request.api_response_validator)
        sql_list = parse_api_field(request.sql)
        sql_str = sql_list[0] if sql_list else None
        if sql_str and isinstance(sql_str, str):
            sql_str = sql_str.strip() or None

        # Same routing flags as multi-agent download (after question_id may populate SQL/API).
        has_api = bool(
            request.api_endpoint
            and request.api_params
            and request.api_response_validator
            and (request.api_endpoint != "N" and request.api_endpoint != "[]")
        )
        _sql_raw_dl = getattr(request, "sql", None)
        has_sql = bool(
            _sql_raw_dl
            and (
                (isinstance(_sql_raw_dl, list) and _sql_raw_dl)
                or (
                    isinstance(_sql_raw_dl, str)
                    and _sql_raw_dl.strip()
                    and _sql_raw_dl != "N"
                )
            )
        )

        # @askme before API/SQL only when there is no saved SQL/API to execute first (request.df /
        # standalone resolve). When question_id loads SQL/API, we run those branches first, then
        # @askme on the executed df (same pattern as multi_agent_download_excel).
        askme_handled = False
        if (
            "@askme" in (request.my_question or "").lower()
            and getattr(settings, "IS_ASKME_ACTIVE", False)
            and not has_api
            and not has_sql
        ):
            prior_input, prior_err = await _resolve_prior_df_input_for_askme(
                service, request, single_agent_id
            )
            if prior_err:
                return {"success": False, "message": prior_err, "details": {}}
            if (request.text_analysis or "").strip() or (request.text_analysis_html or "").strip():
                summary, df_display, fig, askme_err = _askme_excel_from_cached_ask_response(
                    service, request, prior_input
                )
            else:
                summary, df_display, fig, askme_err = await _askme_excel_payload(
                    service,
                    request.my_question,
                    prior_input,
                    request.code,
                )
            if askme_err:
                return {"success": False, "message": askme_err, "details": {}}
            df = None
            askme_handled = True

        # Handle API flow: reuse saved/request API when present (no re-discovery)
        if not askme_handled and api_endpoints:
            print("[DOWNLOAD-EXCEL] Reusing saved API endpoint and params")
            try:
                # If we have multiple endpoints, use the multi-API merge path
                if len(api_endpoints) > 1:
                    print(f"[DOWNLOAD-EXCEL] Multi-API detected for single agent: {api_endpoints}")
                    
                    # Get all APIs for this agent
                    all_apis = await run_in_threadpool(service.get_all_apis, single_agent_id)
                    
                    # Resolve specified endpoints to API configurations
                    selected_apis = []
                    for ep in api_endpoints:
                        for api in all_apis:
                            if api.get("endpoint") == ep:
                                selected_apis.append(api)
                                break
                    
                    if not selected_apis:
                        return {
                            "success": False,
                            "message": f"API endpoints {api_endpoints} not found for agent '{single_agent_id}'",
                            "details": {}
                        }
                    
                    # Execute multi-API call
                    svc_api = service._ensure_api_chat_service()
                    merge_result = await run_in_threadpool(
                        svc_api.call_multiple_apis_and_merge,
                        selected_apis=selected_apis,
                        user_query=request.my_question
                    )
                    
                    if not merge_result.get("success"):
                        return {
                            "success": False,
                            "message": merge_result.get("message", "Failed to retrieve data from multiple APIs"),
                            "details": {}
                        }
                    
                    df = merge_result.get("df")

                else:
                    # Single API endpoint (optimized path)
                    api_endpoint_str = api_endpoints[0]
                    api_params_str = api_params_list[0] if api_params_list else ""
                    api_response_validator_str = api_validators[0] if api_validators else ""

                    print(f"[DOWNLOAD-EXCEL] Single-API call: {api_endpoint_str}")
                    
                    # Parse the params (usually stored as a string representation of a dict)
                    try:
                        extracted_params = ast.literal_eval(api_params_str) if isinstance(api_params_str, str) and api_params_str.strip() else {}
                        if not isinstance(extracted_params, dict):
                            extracted_params = {}
                    except Exception:
                        extracted_params = {}
                    
                    apis_list = await run_in_threadpool(service.get_all_apis, single_agent_id)
                    api_info = {}
                    for api in apis_list:
                        if api.get("endpoint") == api_endpoint_str:
                            api_info = api
                            break
                    
                    if api_info:
                        is_sap = api_info.get("extra1", "N")
                        
                        if is_sap == "Y":
                            # SAP Flow
                            sap_result = await run_in_threadpool(
                                service.process_sap_query,
                                service_root=api_endpoint_str,
                                entity_set=api_info.get("base_params", ""),
                                user_query=request.my_question,
                                response_schema=api_info.get("response_schema", ""),
                                transformation_code=api_response_validator_str or "",
                                username="S4h_aa",
                                password="Welcome1",
                                verify_ssl=False,
                                api_query_params=extracted_params
                            )
                            df = sap_result.get("df")
                        else:
                            # Non-SAP API Flow
                            _data, _data2, df, _data3 = await run_in_threadpool(
                                service.call_api_and_transform,
                                extracted_params,
                                api_endpoint_str,
                                api_info.get("response_schema", ""),
                                api_response_validator_str or "",
                                request.my_question
                            )
                    else:
                        return {
                            "success": False,
                            "message": f"API endpoint '{api_endpoint_str}' not found for agent '{single_agent_id}'",
                            "details": {}
                        }
            except Exception as e:
                print(f"Error calling API: {e}")
                return {
                    "success": False,
                    "message": f"Error calling API: {str(e)}",
                    "details": {}
                }
            if not summary:
                summary = _plain_text_from_sql_query_summaries(
                    getattr(request, "sql_query_summaries", None)
                ) or summary
        # Handle Database flow: reuse saved/request SQL when present (no regeneration)
        elif not askme_handled and sql_str and sql_str != "N":
            print("[DOWNLOAD-EXCEL] Reusing saved SQL")
            service._ensure_sql()
            sql_to_use = sql_str
            try:
                if service.is_sql_valid_cached(sql=sql_to_use):
                    df = service.run_sql_cached(sql=sql_to_use)
                else:
                    return {
                        "success": False,
                        "message": (
                            "We could not use that as a valid query. Rephrase your question, or use one of "
                            "the suggested questions."
                        ),
                        "details": {}
                    }
            except Exception as e:
                print(f"Error running SQL: {e}")
                return {
                    "success": False,
                    "message": f"Error running SQL: {str(e)}",
                    "details": {}
                }
            if not summary:
                summary = _plain_text_from_sql_query_summaries(
                    getattr(request, "sql_query_summaries", None)
                ) or summary
        # @web — same pipeline as ask-question (Tavily + optional prior df from request.df)
        elif not askme_handled and "@web" in (request.my_question or "").lower() and getattr(
            settings, "IS_WEB_ACTIVE", False
        ):
            summary, df_display, fig, web_err = await _web_excel_payload(
                service,
                request.my_question,
                getattr(request, "df", None),
                request.code,
            )
            if web_err:
                return {"success": False, "message": web_err, "details": {}}
            df = None  # not used; df_display already indexed for Excel
        # Default flow - only when no API and no SQL: generate SQL once
        elif not askme_handled:
            service._ensure_sql()
            sql_to_use = service.generate_sql_cached(question=request.my_question)
            if sql_to_use and sql_to_use != "N":
                try:
                    if service.is_sql_valid_cached(sql=sql_to_use):
                        df = service.run_sql_cached(sql=sql_to_use)
                except Exception as e:
                    print(f"Error in default flow: {e}")

        # @askme after saved/API or SQL execution: analyze the same in-memory df (question_id path).
        if (
            not askme_handled
            and getattr(settings, "IS_ASKME_ACTIVE", False)
            and "@askme" in (request.my_question or "").lower()
        ):
            executed_prior = _prior_input_from_executed_df(df)
            if executed_prior is not None:
                prior_input, prior_err = executed_prior, None
            else:
                prior_input, prior_err = await _resolve_prior_df_input_for_askme(
                    service, request, single_agent_id
                )
            if prior_err:
                return {"success": False, "message": prior_err, "details": {}}
            if (request.text_analysis or "").strip() or (request.text_analysis_html or "").strip():
                summary, df_display, fig, askme_err = _askme_excel_from_cached_ask_response(
                    service, request, prior_input
                )
            else:
                summary, df_display, fig, askme_err = await _askme_excel_payload(
                    service,
                    request.my_question,
                    prior_input,
                    request.code,
                )
            if askme_err:
                return {"success": False, "message": askme_err, "details": {}}
            df = None
            askme_handled = True

        # Process DataFrame if available
        print("json_code", request.code)
        if df is not None and not df.empty:
            # Prepare table display
            df_display = df.copy()
            df_display.index = df_display.index + 1
            
            # Get columns for chart generation
            columns = list(df.columns) if hasattr(df, 'columns') else []
            print("json_code before conversion", request.code)
            # Generate chart if appropriate
            if len(df) <= 100:
                try:
                    # Check if chart should be generated
                    json_code = request.code
                    
                    if json_code and json_code != "N" and json_code != "":
                        # Generate chart from code (matching askme.py pattern)
                        fig = generate_chart_from_code(json_code, df, columns, request.my_question, service)
                        print("converted json_code", fig)
                except Exception as e:
                    import traceback
        
        # Create Excel file
        excel_buffer = create_excel_file(
            question=request.my_question,
            chart_fig=fig,
            data=df_display if df_display is not None else None,
            summary=summary
        )
        
        # Format filename - clean question for filename
        clean_question = "".join([c if c.isalnum() else "_" for c in request.my_question])[:30]
        timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
        file_id = str(uuid.uuid4())[:8]
        filename = f"askme360_response_{clean_question}_{timestamp}_{file_id}.xlsx"
        
        # Read the buffer content
        excel_bytes = excel_buffer.getvalue()
        
        # Create downloads directory if it doesn't exist
        downloads_dir = os.path.join(os.getcwd(), "downloads")
        os.makedirs(downloads_dir, exist_ok=True)
        
        # Save file to disk
        file_path = os.path.join(downloads_dir, filename)
        with open(file_path, 'wb') as f:
            f.write(excel_bytes)
        
        # Handle email sending if requested
        email_sent = False
        email_error = None
        email_recipients = None
        
        if request.sendemail == "Y" and request.email_list:
            # Also save to email_files directory for email attachment
            email_dir = os.path.join(os.getcwd(), "email_files")
            os.makedirs(email_dir, exist_ok=True)
            email_file_path = os.path.join(email_dir, filename)
            
            # Write bytes to email file
            with open(email_file_path, 'wb') as f:
                f.write(excel_bytes)
            
            # Get email content
            try:
                body_html, subject = get_mail_content(
                    schedule_type=None,  # Regular email, not scheduled
                    activity=request.my_question,
                    description="Please find attached the requested report."
                )
                
                # Send email
                email_sent = service.send_email(
                    sender_email=agent_config.get("sender_email"),
                    sender_name="askme360",
                    recipient_email=request.email_list,
                    subject=subject,
                    body_text="Report of the:",
                    body_html=body_html,
                    attachment_path=email_file_path,
                    agent_conf=agent_config
                )
                
                if email_sent:
                    email_recipients = request.email_list
                else:
                    email_error = "Failed to send email. Please check SMTP configuration."
                    
            except Exception as e:
                email_error = f"Error sending email: {str(e)}"
                print(f"[DOWNLOAD-EXCEL] Email error: {email_error}")
        
        # Return JSON response with download link (absolute URL so the client can open it directly)
        base_url = str(http_request.base_url).rstrip("/")
        download_link = f"/api/download/file/{filename}"
        
        response_details = {
            "link": download_link,
            "filename": filename,
            "file_path": file_path,
            "email_sent": email_sent
        }
        
        if email_recipients:
            response_details["email_recipients"] = email_recipients
        
        if email_error:
            response_details["email_error"] = email_error
        
        return {
            "success": True,
            "message": "Excel file created successfully",
            "details": response_details
        }
    
    except Exception as e:
        return {
            "success": False,
            "message": f"Error creating Excel file: {str(e)}",
            "details": {}
        }


@router.post("/multi-agent-download-excel")
async def multi_agent_download_excel(
    request: DownloadExcelRequest,
    http_request: Request,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """
    Download Excel file with question response data for single-agent OR multi-agent requests.

    Accepts the same DownloadExcelRequest schema as /download-excel.
    When agent_id is a list, the endpoint fans out across agents:
      - API datasource: collects APIs from all agents, selects the best match,
        calls and merges results, then builds the Excel.
      - Database datasource: uses the first (primary) agent for SQL generation
        and execution, same as the single-agent path.
    When agent_id is a single string the behaviour is identical to /download-excel.

    Body:
        my_question: The question text (required)
        agent_id: Single agent ID string OR list of agent IDs (required)
        sendemail: Send email flag - "Y" or "N" (required)
        email_list: List of email addresses (optional)
        sql: SQL query (optional)
        code: Chart code (optional)
        api_endpoint: API endpoint (optional)
        api_params: API parameters (optional)
        api_response_validator: API response validator (optional)
        question_id: Question ID to fetch saved data (optional)
        df: Prior table for ``@askme`` (required) / optional for ``@web`` (same as ask-question)

    Returns:
        Excel file download response
    """
    import json as _json

    try:
        # ── Normalize agent_id to a list ─────────────────────────────────────
        raw_agent_id = request.agent_id
        if isinstance(raw_agent_id, list):
            agent_ids = raw_agent_id
        elif isinstance(raw_agent_id, str) and raw_agent_id.startswith('[') and raw_agent_id.endswith(']'):
            try:
                agent_ids = _json.loads(raw_agent_id)
                if not isinstance(agent_ids, list):
                    agent_ids = [raw_agent_id]
            except Exception:
                agent_ids = [raw_agent_id]
        elif isinstance(raw_agent_id, str):
            agent_ids = [raw_agent_id]
        else:
            agent_ids = list(raw_agent_id) if raw_agent_id else []

        if not agent_ids:
            return {
                "success": False,
                "message": "No agent IDs provided",
                "details": {}
            }

        primary_agent_id = agent_ids[0]  # Always use first agent as primary
        is_multi_agent = len(agent_ids) > 1

        # ── If question_id provided, fetch saved question and reuse SQL/API (same as multi-agent-ask-question)
        if request.question_id:
            try:
                user_id = str(getattr(current_user, "id", "")) or getattr(current_user, "user_id", "")
                question = db.query(Question).filter(
                    Question.id == int(request.question_id),
                    Question.user_id == user_id,
                ).first()
                if question:
                    request.my_question = question.question or request.my_question
                    request.sql = question.sql or request.sql
                    request.code = question.chart or request.code
                    request.api_endpoint = question.api_endpoint or request.api_endpoint
                    request.api_params = question.api_params or request.api_params
                    request.api_response_validator = question.api_response_validator or request.api_response_validator
                    request.sql_query_summaries = (
                        getattr(question, "sql_query_summaries", None)
                        or request.sql_query_summaries
                    )
                else:
                    return {
                        "success": False,
                        "message": f"Question not found for question_id: {request.question_id}",
                        "details": {}
                    }
            except Exception as e:
                print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Error fetching question data: {e}")
                return {
                    "success": False,
                    "message": f"Error fetching question data: {str(e)}",
                    "details": {}
                }
        else:
            try:
                user_id = str(getattr(current_user, "id", "")) or getattr(current_user, "user_id", "")
                await run_in_threadpool(_hydrate_request_from_user_question_log_cache, request, user_id)
            except Exception:
                pass

        # ── Get primary agent config & service ───────────────────────────────
        agent_config = get_agent_config(agent_id=primary_agent_id)
        if not agent_config:
            return {
                "success": False,
                "message": f"Agent configuration not found for agent_id: {primary_agent_id}",
                "details": {}
            }

        primary_service = get_askme_service(agent_id=primary_agent_id)
        if not primary_service:
            return {
                "success": False,
                "message": f"Failed to initialize service for agent '{primary_agent_id}'",
                "details": {}
            }

        if _is_document_question(
            _resolve_download_question_text(request),
            getattr(request, "document_result_type", None),
            getattr(request, "is_file", None),
        ):
            if request.sendemail == "Y" and request.email_list:
                return await _process_document_email_from_download_request(
                    request,
                    primary_agent_id,
                    agent_config,
                    primary_service,
                )
            return {
                "success": False,
                "message": "Excel export is not supported for document questions.",
                "details": {},
            }

        # Parse API fields robustly (trim trailing "" slots from JSON arrays)
        api_endpoints, api_params_list, api_validators = _trim_trailing_empty_api_slots(
            parse_api_field(request.api_endpoint),
            parse_api_field(request.api_params),
            parse_api_field(request.api_response_validator),
        )

        df = None
        fig = None
        df_display = None
        summary = None
        merge_result: Optional[Dict[str, Any]] = None
        mixed_download_handled = False
        askme_handled = False
        db_frames: List[pd.DataFrame] = []
        db_labels: List[str] = []
        api_frames: List[pd.DataFrame] = []
        api_labels: List[str] = []

        # ── Routing: reuse saved/request SQL or API when present (no regeneration)
        has_api = bool(
            request.api_endpoint and request.api_params and request.api_response_validator
            and (request.api_endpoint != "N" and request.api_endpoint != "[]")
        )
        _sql_raw = getattr(request, "sql", None)
        has_sql = bool(
            _sql_raw
            and ((isinstance(_sql_raw, list) and _sql_raw) or (isinstance(_sql_raw, str) and _sql_raw.strip() and _sql_raw != "N"))
        )

        # Mixed DB+API (multi-agent): for download we keep DB/API separate sheets.
        # We intentionally avoid mixed merged DataFrame execution in this endpoint.
        if (
            is_multi_agent
            and has_sql
            and has_api
            and _download_meaningful_sql(_sql_raw)
            and _download_meaningful_api(request.api_endpoint, request.api_params)
        ):
            print(
                "[MULTI-AGENT-DOWNLOAD-EXCEL] Both SQL and API provided; "
                "running independent SQL and API flows for separate sheets"
            )

        if (
            not mixed_download_handled
            and "@askme" in (request.my_question or "").lower()
            and getattr(settings, "IS_ASKME_ACTIVE", False)
            and not has_api
            and not has_sql
        ):
            prior_input, prior_err = await _resolve_prior_df_input_for_askme(
                primary_service, request, primary_agent_id
            )
            if prior_err:
                return {"success": False, "message": prior_err, "details": {}}
            if (request.text_analysis or "").strip() or (request.text_analysis_html or "").strip():
                summary, df_display, fig, askme_err = _askme_excel_from_cached_ask_response(
                    primary_service, request, prior_input
                )
            else:
                summary, df_display, fig, askme_err = await _askme_excel_payload(
                    primary_service,
                    request.my_question,
                    prior_input,
                    request.code,
                )
            if askme_err:
                return {"success": False, "message": askme_err, "details": {}}
            df = None
            askme_handled = True

        if not mixed_download_handled and not askme_handled and has_api:
            # ── API FLOW - Reusing saved/request API ───────────────────────────
            try:
                api_df_out = None
                print("[MULTI-AGENT-DOWNLOAD-EXCEL] Reusing saved API endpoint and params")
                if is_multi_agent:
                    # Collect ALL APIs from every agent
                    all_apis = []
                    for aid in agent_ids:
                        svc = get_askme_service(agent_id=aid)
                        if svc:
                            apis = svc.get_all_apis(aid)
                            if apis:
                                all_apis.extend(apis)
                    print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Collected {len(all_apis)} APIs across {len(agent_ids)} agents")

                    if not all_apis:
                        if not (has_sql and _download_meaningful_sql(_sql_raw)):
                            return {
                                "success": False,
                                "message": "No APIs configured across the selected agents",
                                "details": {}
                            }
                    else:
                        selected_apis = []
                        for ep in api_endpoints:
                            for api in all_apis:
                                if api.get("endpoint") == ep:
                                    selected_apis.append(api)
                                    break

                        if not selected_apis:
                            return {
                                "success": False,
                                "message": f"API endpoints '{api_endpoints}' not found across selected agents",
                                "details": {}
                            }

                        print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Selected APIs: {[a.get('name') for a in selected_apis]}")

                        # Ensure API chat service exists
                        svc = primary_service._ensure_api_chat_service()

                        merge_result = await run_in_threadpool(
                            svc.call_multiple_apis_and_merge,
                            selected_apis=selected_apis,
                            user_query=request.my_question
                        )

                        if not merge_result.get("success"):
                            return {
                                "success": False,
                                "message": merge_result.get("message", "Failed to merge data from APIs"),
                                "details": {}
                            }

                        api_df_out = merge_result.get("df")

                else:
                    # Single-agent branch (align with download_excel)
                    if len(api_endpoints) > 1:
                        print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Multi-API detected for single agent: {api_endpoints}")

                        all_apis = await run_in_threadpool(primary_service.get_all_apis, primary_agent_id)
                        selected_apis = []
                        for ep in api_endpoints:
                            for api in all_apis:
                                if api.get("endpoint") == ep:
                                    selected_apis.append(api)
                                    break

                        if not selected_apis:
                            return {
                                "success": False,
                                "message": f"API endpoints {api_endpoints} not found for agent '{primary_agent_id}'",
                                "details": {}
                            }

                        svc_api = primary_service._ensure_api_chat_service()
                        merge_result = await run_in_threadpool(
                            svc_api.call_multiple_apis_and_merge,
                            selected_apis=selected_apis,
                            user_query=request.my_question
                        )

                        if not merge_result.get("success"):
                            return {
                                "success": False,
                                "message": merge_result.get("message", "Failed to retrieve data from multiple APIs"),
                                "details": {}
                            }

                        api_df_out = merge_result.get("df")
                    else:
                        # Single API endpoint (optimized path)
                        api_endpoint_str = api_endpoints[0]
                        api_params_str = api_params_list[0] if api_params_list else ""
                        api_response_validator_str = api_validators[0] if api_validators else ""

                        print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Single-API call: {api_endpoint_str}")

                        try:
                            extracted_params = ast.literal_eval(api_params_str) if isinstance(api_params_str, str) and api_params_str.strip() else {}
                            if not isinstance(extracted_params, dict):
                                extracted_params = {}
                        except Exception:
                            extracted_params = {}

                        apis_list = await run_in_threadpool(primary_service.get_all_apis, primary_agent_id)
                        api_info = {}
                        for api in apis_list:
                            if api.get("endpoint") == api_endpoint_str:
                                api_info = api
                                break

                        if api_info:
                            is_sap = api_info.get("extra1", "N")
                            if is_sap == "Y":
                                sap_result = await run_in_threadpool(
                                    primary_service.process_sap_query,
                                    service_root=api_endpoint_str,
                                    entity_set=api_info.get("base_params", ""),
                                    user_query=request.my_question,
                                    response_schema=api_info.get("response_schema", ""),
                                    transformation_code=api_response_validator_str or "",
                                    username="S4h_aa",
                                    password="Welcome1",
                                    verify_ssl=False,
                                    api_query_params=extracted_params
                                )
                                api_df_out = sap_result.get("df")
                            else:
                                _data, _data2, api_df_out, _data3 = await run_in_threadpool(
                                    primary_service.call_api_and_transform,
                                    extracted_params,
                                    api_endpoint_str,
                                    api_info.get("response_schema", ""),
                                    api_response_validator_str or "",
                                    request.my_question
                                )
                        else:
                            return {
                                "success": False,
                                "message": f"API endpoint '{api_endpoint_str}' not found for agent '{primary_agent_id}'",
                                "details": {}
                            }

                api_details = (merge_result or {}).get("api_details", []) if isinstance(merge_result, dict) else []
                api_names = [str(d.get("name", "")).strip() for d in api_details if isinstance(d, dict)]
                api_frames_found = _collect_non_empty_frames(api_df_out)
                for idx, frame in enumerate(api_frames_found):
                    api_frames.append(frame)
                    if idx < len(api_names) and api_names[idx]:
                        api_labels.append(api_names[idx])
                    else:
                        api_labels.append(f"API View {len(api_labels) + 1}")
                if api_frames_found:
                    df = api_df_out

            except Exception as e:
                print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Error calling API: {e}")
                return {
                    "success": False,
                    "message": f"Error calling API: {str(e)}",
                    "details": {}
                }
            if not summary:
                summary = _plain_text_from_sql_query_summaries(
                    getattr(request, "sql_query_summaries", None)
                ) or summary

        if not mixed_download_handled and not askme_handled and has_sql:
            # ── DATABASE FLOW - Reusing saved/request SQL ─────────────────────
            print("[MULTI-AGENT-DOWNLOAD-EXCEL] Reusing saved SQL")
            # Multi-agent: use same logic as multi-agent-ask-question (per-agent SQL, merge)
            sql_raw = request.sql
            is_multi_agent_sql = False
            merge_info = None
            sql_for_run = None

            if is_multi_agent:
                if isinstance(sql_raw, list):
                    # List of SQLs: one per agent
                    sql_list = list(sql_raw)[:len(agent_ids)]
                    if len(sql_list) < len(agent_ids):
                        pad = sql_list[-1] if sql_list else ""
                        sql_list = sql_list + [pad] * (len(agent_ids) - len(sql_list))
                    merge_info = {
                        "queries": {str(agent_ids[i]): sql_list[i] for i in range(len(agent_ids))},
                        "execution_order": list(agent_ids),
                        "merge_key": "id",
                        "merge_strategy": "independent",
                    }
                    sql_for_run = json.dumps(merge_info)
                    is_multi_agent_sql = True
                elif isinstance(sql_raw, (str, dict)):
                    parsed_merge = _download_merge_info_from_sql_blob(sql_raw)
                    if parsed_merge and parsed_merge.get("queries"):
                        merge_info = parsed_merge
                        sql_for_run = json.dumps(merge_info)
                        is_multi_agent_sql = True

            if is_multi_agent_sql and merge_info and merge_info.get("queries"):
                try:
                    multi_agent_service = get_askme_service(agent_id=agent_ids)
                    if not multi_agent_service:
                        return {
                            "success": False,
                            "message": "Failed to initialize multi-agent service",
                            "details": {}
                        }
                    multi_agent_service._ensure_sql()
                    sql_df_out = multi_agent_service.run_sql_cached(
                        sql_for_run, merge_info=merge_info, question=request.my_question
                    )
                    if isinstance(sql_df_out, dict):
                        # Keep as separate tables (same as API datasource): list of dfs → Excel gets two tables
                        execution_order = merge_info.get("execution_order") or list(sql_df_out.keys())
                        df_list = [
                            sql_df_out[aid] for aid in execution_order
                            if sql_df_out.get(aid) is not None and hasattr(sql_df_out.get(aid), "empty") and not sql_df_out[aid].empty
                        ]
                        sql_df_out = df_list if df_list else pd.DataFrame()
                    sql_frames_found = _collect_non_empty_frames(
                        sql_df_out,
                        order=(merge_info.get("execution_order") if merge_info else None),
                    )
                    for idx, frame in enumerate(sql_frames_found):
                        db_frames.append(frame)
                        if merge_info and isinstance(merge_info.get("execution_order"), list) and idx < len(merge_info.get("execution_order", [])):
                            db_labels.append(f"Agent {merge_info.get('execution_order')[idx]}")
                        elif idx < len(agent_ids):
                            db_labels.append(f"Agent {agent_ids[idx]}")
                        else:
                            db_labels.append(f"DB View {len(db_labels) + 1}")
                    if sql_frames_found:
                        df = sql_df_out
                except Exception as e:
                    print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Error running multi-agent SQL: {e}")
                    return {
                        "success": False,
                        "message": f"Error running SQL: {str(e)}",
                        "details": {}
                    }
            else:
                # Single-agent or single SQL string
                primary_service._ensure_sql()
                sql_to_use = request.sql[0] if isinstance(request.sql, list) else request.sql
                try:
                    if primary_service.is_sql_valid_cached(sql=sql_to_use):
                        sql_df_out = primary_service.run_sql_cached(sql_to_use)
                        sql_frames_found = _collect_non_empty_frames(sql_df_out)
                        for frame in sql_frames_found:
                            db_frames.append(frame)
                            db_labels.append(f"Agent {primary_agent_id}")
                        if sql_frames_found:
                            df = sql_df_out
                    else:
                        return {
                            "success": False,
                            "message": "Provided SQL is not valid",
                            "details": {}
                        }
                except Exception as e:
                    print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Error running SQL: {e}")
                    return {
                        "success": False,
                        "message": f"Error running SQL: {str(e)}",
                        "details": {}
                    }
            if not summary:
                summary = _plain_text_from_sql_query_summaries(
                    getattr(request, "sql_query_summaries", None)
                ) or summary

        if (
            not mixed_download_handled
            and not askme_handled
            and not has_api
            and not has_sql
        ):
            # ── DATABASE FLOW (no API, no SQL: generate SQL once) ─────────────
            if is_multi_agent:
                try:
                    multi_agent_service = get_askme_service(agent_id=agent_ids)
                    if not multi_agent_service:
                        return {
                            "success": False,
                            "message": "Failed to initialize multi-agent service",
                            "details": {}
                        }
                    multi_agent_service._ensure_sql()
                    generated_sql = multi_agent_service.generate_sql_cached(question=request.my_question)
                    if not generated_sql or generated_sql == "N":
                        return {
                            "success": False,
                            "message": "Failed to generate SQL for the question",
                            "details": {}
                        }
                    merge_info = _download_merge_info_from_sql_blob(generated_sql)
                    if merge_info and merge_info.get("queries"):
                        sql_for_run = json.dumps(merge_info)
                        sql_df_out = multi_agent_service.run_sql_cached(
                            sql_for_run, merge_info=merge_info, question=request.my_question
                        )
                        if isinstance(sql_df_out, dict):
                            # Keep as separate tables (same as API datasource): list of dfs → Excel gets two tables
                            execution_order = merge_info.get("execution_order") or list(sql_df_out.keys())
                            df_list = [
                                sql_df_out[aid] for aid in execution_order
                                if sql_df_out.get(aid) is not None and hasattr(sql_df_out.get(aid), "empty") and not sql_df_out[aid].empty
                            ]
                            sql_df_out = df_list if df_list else None
                        sql_frames_found = _collect_non_empty_frames(
                            sql_df_out,
                            order=(merge_info.get("execution_order") if merge_info else None),
                        )
                        for idx, frame in enumerate(sql_frames_found):
                            db_frames.append(frame)
                            if merge_info and isinstance(merge_info.get("execution_order"), list) and idx < len(merge_info.get("execution_order", [])):
                                db_labels.append(f"Agent {merge_info.get('execution_order')[idx]}")
                            elif idx < len(agent_ids):
                                db_labels.append(f"Agent {agent_ids[idx]}")
                            else:
                                db_labels.append(f"DB View {len(db_labels) + 1}")
                        if sql_frames_found:
                            df = sql_df_out
                    else:
                        df = None
                except Exception as e:
                    print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Error in multi-agent auto-SQL flow: {e}")
                    df = None
            else:
                primary_service._ensure_sql()
                sql_to_use = primary_service.generate_sql_cached(question=request.my_question)
                if sql_to_use and sql_to_use != "N":
                    try:
                        if primary_service.is_sql_valid_cached(sql=sql_to_use):
                            sql_df_out = primary_service.run_sql_cached(sql_to_use)
                            sql_frames_found = _collect_non_empty_frames(sql_df_out)
                            for frame in sql_frames_found:
                                db_frames.append(frame)
                                db_labels.append(f"Agent {primary_agent_id}")
                            if sql_frames_found:
                                df = sql_df_out
                    except Exception as e:
                        print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Error in auto-SQL flow: {e}")
        if (
            not askme_handled
            and getattr(settings, "IS_ASKME_ACTIVE", False)
            and "@askme" in (request.my_question or "").lower()
        ):
            executed_prior = _prior_input_from_executed_df(df)
            if executed_prior is not None:
                prior_input, prior_err = executed_prior, None
            else:
                prior_input, prior_err = await _resolve_prior_df_input_for_askme(
                    primary_service, request, primary_agent_id
                )
            if prior_err:
                return {"success": False, "message": prior_err, "details": {}}
            if (request.text_analysis or "").strip() or (request.text_analysis_html or "").strip():
                summary, df_display, fig, askme_err = _askme_excel_from_cached_ask_response(
                    primary_service, request, prior_input
                )
            else:
                summary, df_display, fig, askme_err = await _askme_excel_payload(
                    primary_service,
                    request.my_question,
                    prior_input,
                    request.code,
                )
            if askme_err:
                return {"success": False, "message": askme_err, "details": {}}
            df = None
            askme_handled = True

        # ── Process DataFrame ────────────────────────────────────────────────
        print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] chart code: {request.code}")

        if not askme_handled:
            # Build a fallback `df` for @askme prior-input compatibility if needed.
            if df is None:
                combined = db_frames + api_frames
                if len(combined) == 1:
                    df = combined[0]
                elif len(combined) > 1:
                    df = combined

            chart_df = None
            if db_frames:
                chart_df = db_frames[0]
            elif api_frames:
                chart_df = api_frames[0]
            if chart_df is not None and len(chart_df) <= 100:
                try:
                    columns = list(chart_df.columns) if hasattr(chart_df, "columns") else []
                    json_code = request.code
                    if json_code and json_code != "N" and json_code != "":
                        fig = generate_chart_from_code(
                            json_code, chart_df, columns, request.my_question, primary_service
                        )
                        print(
                            f"[MULTI-AGENT-DOWNLOAD-EXCEL] chart generated: {fig is not None}"
                        )
                except Exception as e:
                    print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Chart error: {e}")

            db_sections = [
                (db_labels[idx] if idx < len(db_labels) else f"DB View {idx + 1}", frame)
                for idx, frame in enumerate(db_frames)
                if frame is not None and not frame.empty
            ]
            api_sections = [
                (api_labels[idx] if idx < len(api_labels) else f"API View {idx + 1}", frame)
                for idx, frame in enumerate(api_frames)
                if frame is not None and not frame.empty
            ]

            if not db_sections and not api_sections:
                return {
                    "success": False,
                    "message": "No tabular data available to export",
                    "details": {},
                }

            excel_buffer = create_excel_file_by_source(
                question=request.my_question,
                db_sections=db_sections,
                api_sections=api_sections,
                chart_fig=fig,
                summary=summary,
            )
        else:
            excel_buffer = create_excel_file(
                question=request.my_question,
                chart_fig=fig,
                data=df_display if df_display is not None else None,
                summary=summary,
                agent_names=None,
            )

        # ── Build filename ────────────────────────────────────────────────────
        clean_question = "".join([c if c.isalnum() else "_" for c in request.my_question])[:30]
        timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
        file_id = str(uuid.uuid4())[:8]
        filename = f"askme360_response_{clean_question}_{timestamp}_{file_id}.xlsx"

        # ── Save to disk ──────────────────────────────────────────────────────
        excel_bytes = excel_buffer.getvalue()
        downloads_dir = os.path.join(os.getcwd(), "downloads")
        os.makedirs(downloads_dir, exist_ok=True)
        file_path = os.path.join(downloads_dir, filename)
        with open(file_path, 'wb') as f:
            f.write(excel_bytes)

        # ── Optional email ────────────────────────────────────────────────────
        email_sent = False
        email_error = None
        email_recipients = None

        if request.sendemail == "Y" and request.email_list:
            email_dir = os.path.join(os.getcwd(), "email_files")
            os.makedirs(email_dir, exist_ok=True)
            email_file_path = os.path.join(email_dir, filename)
            with open(email_file_path, 'wb') as f:
                f.write(excel_bytes)

            try:
                body_html, subject = get_mail_content(
                    schedule_type=None,
                    activity=request.my_question,
                    description="Please find attached the requested report."
                )
                email_sent = primary_service.send_email(
                    sender_email=agent_config.get("sender_email"),
                    sender_name="askme360",
                    recipient_email=request.email_list,
                    subject=subject,
                    body_text="Report of the:",
                    body_html=body_html,
                    attachment_path=email_file_path,
                    agent_conf=agent_config
                )
                if email_sent:
                    email_recipients = request.email_list
                else:
                    email_error = "Failed to send email. Please check SMTP configuration."
            except Exception as e:
                email_error = f"Error sending email: {str(e)}"
                print(f"[MULTI-AGENT-DOWNLOAD-EXCEL] Email error: {email_error}")

        # ── Return response ───────────────────────────────────────────────────
        base_url = str(http_request.base_url).rstrip("/")
        download_link = f"/api/download/file/{filename}"
        response_details = {
            "link": download_link,
            "filename": filename,
            "file_path": file_path,
            "email_sent": email_sent,
            "agent_ids": agent_ids,
            "is_multi_agent": is_multi_agent
        }
        if email_recipients:
            response_details["email_recipients"] = email_recipients
        if email_error:
            response_details["email_error"] = email_error

        return {
            "success": True,
            "message": "Excel file created successfully",
            "details": response_details
        }

    except Exception as e:
        return {
            "success": False,
            "message": f"Error creating Excel file: {str(e)}",
            "details": {}
        }


@router.get("/file/{filename}")
async def download_file(
    filename: str,
    current_user: User = Depends(get_current_user)
):
    """
    Download the Excel file by filename.
    
    Args:
        filename: The name of the file to download
    
    Returns:
        File download response
    """
    try:
        downloads_dir = os.path.join(os.getcwd(), "downloads")
        file_path = os.path.join(downloads_dir, filename)
        
        # Check if file exists
        file_exists = await run_in_threadpool(os.path.exists, file_path)
        if not file_exists:
            return {
                "success": False,
                "message": f"File not found: {filename}",
                "details": {}
            }
        
        # Return file as download response
        return FileResponse(
            path=file_path,
            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            filename=filename,
            headers={
                "Content-Disposition": f"attachment; filename={filename}"
            }
        )
    
    except Exception as e:
        return {
            "success": False,
            "message": f"Error downloading file: {str(e)}",
            "details": {}
        }


# ─── Reconcile Download ──────────────────────────────────────────────────────

class DownloadReconcileRequest(BaseModel):
    """Request model for downloading reconcile Excel file"""
    question: str
    agent_id: str
    data: List[Dict[str, Any]]
    columns: List[str]
    reconciliation_report: Dict[str, Any]
    chart: Optional[Dict[str, Any]] = None
    chart_code: Optional[str] = None
    source_counts: Optional[Dict[str, Any]] = None


def create_reconcile_excel(
    question: str,
    data: List[Dict[str, Any]],
    columns: List[str],
    reconciliation_report: Dict[str, Any],
    chart_fig: Optional[Dict[str, Any]] = None,
    source_counts: Optional[Dict[str, Any]] = None,
) -> io.BytesIO:
    """
    Create a single-sheet Excel workbook for reconcile results.
    Layout: Title → Chart → Data table → Source counts → Reconciliation report.
    """
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "Reconcile Report"

    header_fill = PatternFill(start_color="DDDDDD", end_color="DDDDDD", fill_type="solid")

    ws["A1"] = "askme360 – Reconcile Report"
    ws["A1"].font = Font(size=14, bold=True)
    ws["A2"] = f"Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    ws["A4"] = "Question:"
    ws["A4"].font = Font(bold=True)
    ws["A5"] = question

    current_row = 7

    # ── Chart image ───────────────────────────────────────────────────
    if chart_fig:
        # Handle chart_fig being a JSON string instead of a dict
        if isinstance(chart_fig, str):
            try:
                import json as _json
                chart_fig = _json.loads(chart_fig)
                print(f"[create_reconcile_excel] Parsed chart_fig from JSON string")
            except Exception:
                print(f"[create_reconcile_excel] chart_fig is a string but not valid JSON, skipping chart")
                chart_fig = None

        if chart_fig and isinstance(chart_fig, dict):
            print(f"[create_reconcile_excel] chart_fig keys: {list(chart_fig.keys())}, data traces: {len(chart_fig.get('data', []))}")
            ws.cell(row=current_row, column=1, value="Visualization:").font = Font(bold=True)
            current_row += 1
            try:
                img_bytes = generate_chart_image(chart_fig)
                print(f"[create_reconcile_excel] generate_chart_image returned: {type(img_bytes)}, length: {len(img_bytes) if img_bytes else 0}")
                if img_bytes:
                    img_buf = io.BytesIO(img_bytes)
                    img = Image(img_buf)
                    if img.width > 800:
                        ratio = img.height / img.width
                        img.width = 800
                        img.height = int(800 * ratio)
                    ws.add_image(img, f"A{current_row}")
                    current_row += max(1, int(img.height / 20)) + 2
                    print(f"[create_reconcile_excel] Chart image inserted at row A{current_row - max(1, int(img.height / 20)) - 2}")
                else:
                    ws.cell(row=current_row, column=1, value="Chart generation returned no image")
                    current_row += 2
            except Exception as chart_err:
                print(f"[create_reconcile_excel] Chart image failed: {chart_err}")
                import traceback; traceback.print_exc()
                ws.cell(row=current_row, column=1, value=f"Chart image failed: {chart_err}")
                current_row += 2
        else:
            print(f"[create_reconcile_excel] chart_fig is not a valid dict, type: {type(chart_fig)}")

    # ── Merged data table ─────────────────────────────────────────────
    df = pd.DataFrame(data, columns=columns) if data else pd.DataFrame()
    if not df.empty:
        ws.cell(row=current_row, column=1, value="Merged Data:").font = Font(bold=True)
        current_row += 1
        start_row = current_row

        for c, col_name in enumerate(df.columns, 1):
            cell = ws.cell(row=current_row, column=c, value=str(col_name))
            cell.font = Font(bold=True)
            cell.fill = header_fill
            cell.alignment = Alignment(wrap_text=True)
        current_row += 1

        display_df = df.iloc[:1000]
        for _, row_vals in display_df.iterrows():
            for c, val in enumerate(row_vals, 1):
                cell = ws.cell(row=current_row, column=c)
                cell.value = str(val) if isinstance(val, (dict, list)) else val
                cell.alignment = Alignment(wrap_text=True)
            current_row += 1

        # Auto-width
        for c in range(1, len(df.columns) + 1):
            col_letter = get_column_letter(c)
            max_len = len(str(df.columns[c - 1]))
            for r in range(start_row + 1, min(current_row, start_row + 51)):
                cv = ws.cell(row=r, column=c).value
                if cv:
                    max_len = max(max_len, min(len(str(cv)), 50))
            ws.column_dimensions[col_letter].width = max_len + 2

        if len(df) > 1000:
            current_row += 1
            ws.cell(row=current_row, column=1,
                    value=f"Note: Showing first 1,000 of {len(df)} rows").font = Font(italic=True)
        current_row += 2

    # ── Source counts ─────────────────────────────────────────────────
    src = source_counts or {}
    if src:
        ws.cell(row=current_row, column=1, value="Source Summary:").font = Font(bold=True)
        current_row += 1
        for ci, hdr in enumerate(["Source", "Count"], 1):
            cell = ws.cell(row=current_row, column=ci, value=hdr)
            cell.font = Font(bold=True)
            cell.fill = header_fill
        current_row += 1
        for src_key, src_val in src.items():
            ws.cell(row=current_row, column=1, value=str(src_key))
            ws.cell(row=current_row, column=2, value=src_val)
            current_row += 1
        current_row += 2

    # ── Reconciliation report ─────────────────────────────────────────
    report = reconciliation_report or {}

    ws.cell(row=current_row, column=1, value="Reconciliation Report").font = Font(bold=True, size=12)
    current_row += 1

    ws.cell(row=current_row, column=1, value="Status:").font = Font(bold=True)
    ws.cell(row=current_row, column=2, value=report.get("status", "UNKNOWN")).font = Font(bold=True)
    current_row += 1

    ws.cell(row=current_row, column=1, value="Summary:").font = Font(bold=True)
    ws.cell(row=current_row, column=2, value=report.get("summary", ""))
    current_row += 2

    # Column Differences
    col_diffs = report.get("column_differences", {})
    ws.cell(row=current_row, column=1, value="Column Differences:").font = Font(bold=True)
    current_row += 1
    ws.cell(row=current_row, column=1, value="In Excel, not in DB:")
    ws.cell(row=current_row, column=2, value=", ".join(col_diffs.get("in_excel_not_df", [])) or "None")
    current_row += 1
    ws.cell(row=current_row, column=1, value="In DB, not in Excel:")
    ws.cell(row=current_row, column=2, value=", ".join(col_diffs.get("in_df_not_excel", [])) or "None")
    current_row += 2

    # Missing Rows
    missing = report.get("missing_rows", [])
    ws.cell(row=current_row, column=1, value=f"Missing Rows ({len(missing)}):").font = Font(bold=True)
    current_row += 1
    if missing:
        for m in missing[:100]:
            ws.cell(row=current_row, column=1, value=str(m))
            current_row += 1
    else:
        ws.cell(row=current_row, column=1, value="None")
        current_row += 1
    current_row += 1

    # Extra Rows
    extra = report.get("extra_rows", [])
    ws.cell(row=current_row, column=1, value=f"Extra Rows ({len(extra)}):").font = Font(bold=True)
    current_row += 1
    if extra:
        for e in extra[:100]:
            ws.cell(row=current_row, column=1, value=str(e))
            current_row += 1
    else:
        ws.cell(row=current_row, column=1, value="None")
        current_row += 1
    current_row += 1

    # Value Mismatches
    mismatches = report.get("value_mismatches", [])
    ws.cell(row=current_row, column=1, value=f"Value Mismatches ({len(mismatches)}):").font = Font(bold=True)
    current_row += 1
    if mismatches:
        for ci, hdr in enumerate(["Row ID", "Column", "DB Value", "Excel Value"], 1):
            cell = ws.cell(row=current_row, column=ci, value=hdr)
            cell.font = Font(bold=True)
            cell.fill = header_fill
        current_row += 1
        for mm in mismatches[:200]:
            ws.cell(row=current_row, column=1, value=str(mm.get("row_id", "")))
            ws.cell(row=current_row, column=2, value=str(mm.get("column", "")))
            ws.cell(row=current_row, column=3, value=str(mm.get("df_value", "")))
            ws.cell(row=current_row, column=4, value=str(mm.get("excel_value", "")))
            current_row += 1
    else:
        ws.cell(row=current_row, column=1, value="None")
        current_row += 1
    current_row += 1

    # Recommendations
    ws.cell(row=current_row, column=1, value="Recommendations:").font = Font(bold=True)
    current_row += 1
    rec_text = report.get("recommendations", "")
    if rec_text:
        for line in str(rec_text).split("\n"):
            ws.cell(row=current_row, column=1, value=line)
            current_row += 1

    # ── Save ──────────────────────────────────────────────────────────
    buf = io.BytesIO()
    wb.save(buf)
    buf.seek(0)
    return buf


@router.post("/download-reconcile")
def download_reconcile(
    request: DownloadReconcileRequest,
    http_request: Request,
    current_user: User = Depends(get_current_user),
):
    """
    Generate and download an Excel file from a /reconcile API response.

    The request body mirrors the `details` object returned by the /reconcile
    endpoint so the frontend can forward it directly.

    Returns:
        JSON with download link and filename.
    """
    try:
        print(f"[download-reconcile] Received: agent_id={request.agent_id}, chart_code={bool(request.chart_code)}, data_rows={len(request.data)}, columns={len(request.columns)}")

        # Build a clean DataFrame for chart (exclude internal reconciliation columns)
        chart_fig = None
        chart_df = pd.DataFrame(request.data, columns=request.columns)
        chart_cols = [c for c in chart_df.columns if c not in ("_reconciled", "_source")]
        chart_df = chart_df[chart_cols]

        # Determine chart code — use provided chart_code, or generate fresh via LLM
        json_code = request.chart_code
        if not json_code or json_code.strip() == "" or json_code == "N":
            # Generate chart code from LLM — same as download-excel and ask-question
            try:
                service = get_askme_service(agent_id=request.agent_id)
                if service and len(chart_df) > 0 and len(chart_df) < 101 and len(chart_cols) >= 2:
                    json_code = service.generate_json_code_cached(
                        question=request.question,
                        columns=chart_cols,
                        model="gpt-4.1"
                    )
                    print(f"[download-reconcile] LLM generated chart code: {bool(json_code and json_code != 'N')}")
            except Exception as gen_err:
                print(f"[download-reconcile] Chart code generation failed: {gen_err}")
                json_code = None

        # Execute chart code against the actual data — same as download-excel
        if json_code and json_code.strip() and json_code != "N":
            try:
                chart_fig = generate_chart_from_code(
                    code=json_code,
                    df=chart_df,
                    columns=chart_cols,
                    question=request.question,
                    service=None
                )
                print(f"[download-reconcile] Chart generated from code: {bool(chart_fig)}")
            except Exception as chart_err:
                print(f"[download-reconcile] Chart code execution failed: {chart_err}")
                chart_fig = None

        # Fallback to pre-computed chart if code-based generation failed
        if not chart_fig and request.chart:
            chart_fig = request.chart
            print(f"[download-reconcile] Falling back to pre-computed chart")

        excel_buffer = create_reconcile_excel(
            question=request.question,
            data=request.data,
            columns=request.columns,
            reconciliation_report=request.reconciliation_report,
            chart_fig=chart_fig,
            source_counts=request.source_counts,
        )

        # Build filename
        clean_q = "".join(c if c.isalnum() else "_" for c in request.question)[:30]
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        file_id = str(uuid.uuid4())[:8]
        filename = f"askme360_reconcile_{clean_q}_{timestamp}_{file_id}.xlsx"

        # Persist to downloads folder
        downloads_dir = os.path.join(os.getcwd(), "downloads")
        os.makedirs(downloads_dir, exist_ok=True)
        file_path = os.path.join(downloads_dir, filename)
        with open(file_path, "wb") as f:
            f.write(excel_buffer.getvalue())

        base_url = str(http_request.base_url).rstrip("/")
        download_link = f"{base_url}/api/download/file/{filename}"

        return {
            "success": True,
            "message": "Reconcile Excel file created successfully",
            "details": {
                "link": download_link,
                "filename": filename,
                "file_path": file_path,
            },
        }

    except Exception as e:
        import traceback
        traceback.print_exc()
        return {
            "success": False,
            "message": f"Error creating reconcile Excel: {str(e)}",
            "details": {},
        }
