"""
数据产品服务
提供数据产品的列表查询、数据预览、Excel导出、注册等功能
提供数据订单的创建、分析、审批等功能
"""

from __future__ import annotations

import io
import json
import logging
from datetime import datetime
from typing import Any

from flask import current_app
from openai import OpenAI
from sqlalchemy import text

from app import db
from app.models.data_product import DataOrder, DataProduct
from app.services.neo4j_driver import neo4j_driver

logger = logging.getLogger(__name__)


class DataProductService:
    """数据产品服务类"""

    @staticmethod
    def _get_column_tags_from_business_domain(
        product: DataProduct,
    ) -> dict[str, list[dict[str, Any]]]:
        """
        从 Neo4j 获取 BusinessDomain 中列（DataMeta）对应的标签信息

        通过 DataProduct -> DataFlow -> BusinessDomain -> DataMeta -> DataLabel
        的关系链获取每个列对应的标签。

        Args:
            product: 数据产品对象

        Returns:
            列名到标签列表的映射，格式：
            {
                "column_name_en": [
                    {"id": 1, "name_zh": "标签1", "name_en": "tag1"},
                    ...
                ],
                ...
            }
        """
        column_tags: dict[str, list[dict[str, Any]]] = {}

        try:
            with neo4j_driver.get_session() as session:
                bd_id = None

                # 1. 通过 DataFlow 的 OUTPUT 关系找到目标 BusinessDomain
                if product.source_dataflow_id:
                    query = """
                    MATCH (df:DataFlow)-[:OUTPUT]->(bd:BusinessDomain)
                    WHERE id(df) = $dataflow_id
                    RETURN id(bd) as bd_id
                    LIMIT 1
                    """
                    result = session.run(
                        query, {"dataflow_id": product.source_dataflow_id}
                    ).single()
                    if result:
                        bd_id = result["bd_id"]

                # 2. 如果没有找到，尝试通过表名匹配
                if not bd_id:
                    query = """
                    MATCH (bd:BusinessDomain)
                    WHERE bd.name_en = $table_name OR bd.name = $table_name
                    RETURN id(bd) as bd_id
                    LIMIT 1
                    """
                    result = session.run(
                        query, {"table_name": product.target_table}
                    ).single()
                    if result:
                        bd_id = result["bd_id"]

                if not bd_id:
                    logger.debug(f"未找到数据产品关联的BusinessDomain: {product.id}")
                    return column_tags

                # 3. 获取 BusinessDomain 的列（DataMeta）及其标签
                query = """
                MATCH (bd:BusinessDomain)-[inc:INCLUDES]->(m:DataMeta)
                WHERE id(bd) = $bd_id
                OPTIONAL MATCH (m)-[:LABEL]->(label:DataLabel)
                RETURN
                    m.name_en as column_name_en,
                    m.name_zh as column_name_zh,
                    inc.alias_name_en as alias_name_en,
                    inc.alias_name_zh as alias_name_zh,
                    collect(DISTINCT {
                        id: id(label),
                        name_zh: label.name_zh,
                        name_en: label.name_en
                    }) as tags
                """
                result = session.run(query, {"bd_id": bd_id})

                for record in result:
                    # 优先使用别名作为列名（如果有的话）
                    column_key = (
                        record.get("alias_name_en")
                        or record.get("column_name_en")
                        or record.get("alias_name_zh")
                        or record.get("column_name_zh")
                        or ""
                    )

                    if not column_key:
                        continue

                    # 过滤掉空标签（当没有标签关系时会返回 {id: null, ...}）
                    tags = record.get("tags", [])
                    valid_tags = [tag for tag in tags if tag.get("id") is not None]

                    column_tags[column_key] = valid_tags

                    # 同时用中文名作为备用key（如果中英文名不同）
                    column_name_zh = record.get("alias_name_zh") or record.get(
                        "column_name_zh"
                    )
                    if column_name_zh and column_name_zh != column_key:
                        column_tags[column_name_zh] = valid_tags

                logger.debug(f"获取到 {len(column_tags)} 个列的标签信息")

        except Exception as e:
            logger.warning(f"获取列标签信息失败: {str(e)}")

        return column_tags

    @staticmethod
    def get_data_products(
        page: int = 1,
        page_size: int = 20,
        search: str = "",
        status: str | None = None,
    ) -> dict[str, Any]:
        """
        获取数据产品列表

        Args:
            page: 页码
            page_size: 每页大小
            search: 搜索关键词
            status: 状态过滤

        Returns:
            包含数据产品列表和分页信息的字典
        """
        try:
            query = DataProduct.query

            # 搜索过滤
            if search:
                search_pattern = f"%{search}%"
                query = query.filter(
                    db.or_(
                        DataProduct.product_name.ilike(search_pattern),
                        DataProduct.product_name_en.ilike(search_pattern),
                        DataProduct.description.ilike(search_pattern),
                        DataProduct.target_table.ilike(search_pattern),
                    )
                )

            # 状态过滤
            if status:
                query = query.filter(DataProduct.status == status)

            # 计算总数
            total = query.count()

            # 分页查询
            products = (
                query.order_by(DataProduct.created_at.desc())
                .offset((page - 1) * page_size)
                .limit(page_size)
                .all()
            )

            # 转换为字典列表
            product_list = [product.to_dict() for product in products]

            return {
                "list": product_list,
                "pagination": {
                    "page": page,
                    "page_size": page_size,
                    "total": total,
                    "total_pages": (total + page_size - 1) // page_size,
                },
            }

        except Exception as e:
            logger.error(f"获取数据产品列表失败: {str(e)}")
            raise

    @staticmethod
    def get_product_by_id(product_id: int) -> DataProduct | None:
        """
        根据ID获取数据产品

        Args:
            product_id: 数据产品ID

        Returns:
            数据产品对象，不存在则返回None
        """
        return DataProduct.query.get(product_id)

    @staticmethod
    def get_product_preview(
        product_id: int,
        limit: int = 200,
    ) -> dict[str, Any]:
        """
        获取数据产品的数据预览

        Args:
            product_id: 数据产品ID
            limit: 预览数据条数，默认200

        Returns:
            包含列信息和数据的字典
        """
        try:
            product = DataProduct.query.get(product_id)
            if not product:
                raise ValueError(f"数据产品不存在: ID={product_id}")

            # 构建查询SQL
            schema = product.target_schema or "public"
            table = product.target_table
            full_table_name = f"{schema}.{table}"

            # 先检查表是否存在
            check_sql = text(
                """
                SELECT EXISTS (
                    SELECT FROM information_schema.tables
                    WHERE table_schema = :schema
                    AND table_name = :table
                )
                """
            )
            result = db.session.execute(
                check_sql, {"schema": schema, "table": table}
            ).scalar()

            if not result:
                return {
                    "product": product.to_dict(),
                    "columns": [],
                    "data": [],
                    "total_count": 0,
                    "preview_count": 0,
                    "error": f"目标表 {full_table_name} 不存在",
                }

            # 获取列信息
            columns_sql = text(
                """
                SELECT column_name, data_type, is_nullable
                FROM information_schema.columns
                WHERE table_schema = :schema AND table_name = :table
                ORDER BY ordinal_position
                """
            )
            columns_result = db.session.execute(
                columns_sql, {"schema": schema, "table": table}
            ).fetchall()

            columns = [
                {
                    "name": row[0],
                    "type": row[1],
                    "nullable": row[2] == "YES",
                }
                for row in columns_result
            ]

            # 获取 BusinessDomain 中列对应的标签信息
            column_tags = DataProductService._get_column_tags_from_business_domain(
                product
            )

            # 将标签信息合并到 columns 中
            for col in columns:
                col_name = col["name"]
                col["tags"] = column_tags.get(col_name, [])

            # 获取总记录数
            # 使用带引号的表名以避免大小写问题
            if schema == "public":
                count_sql = text(f'SELECT COUNT(*) FROM "{table}"')
            else:
                count_sql = text(f'SELECT COUNT(*) FROM "{schema}"."{table}"')
            try:
                total_count = db.session.execute(count_sql).scalar() or 0
            except Exception as e:
                logger.error(f"查询总记录数失败: {e}, SQL: {count_sql}")
                total_count = 0

            # 获取预览数据
            # 使用带引号的表名以避免大小写问题
            if schema == "public":
                preview_sql = text(f'SELECT * FROM "{table}" LIMIT :limit')
            else:
                preview_sql = text(f'SELECT * FROM "{schema}"."{table}" LIMIT :limit')
            try:
                preview_result = db.session.execute(
                    preview_sql, {"limit": limit}
                ).fetchall()
                logger.debug(f"查询预览数据成功，返回 {len(preview_result)} 行")
            except Exception as e:
                logger.error(f"查询预览数据失败: {e}, SQL: {preview_sql}")
                preview_result = []

            # 转换数据为字典列表
            # 如果从information_schema获取的列信息为空，从查询结果中获取列名
            if columns:
                column_names = [col["name"] for col in columns]
            elif preview_result:
                # 从查询结果的第一行获取列名
                column_names = list(preview_result[0].keys())
                # 同步更新columns列表，包含tags字段
                columns = [
                    {
                        "name": name,
                        "type": "unknown",
                        "nullable": True,
                        "tags": column_tags.get(name, []),
                    }
                    for name in column_names
                ]
            else:
                column_names = []

            data = []
            for row in preview_result:
                # row可能是Row对象或元组
                if hasattr(row, "_mapping"):
                    # SQLAlchemy Row对象（支持列名访问）
                    row_dict = dict(row._mapping)
                elif hasattr(row, "_asdict"):
                    # namedtuple或类似对象
                    row_dict = row._asdict()
                elif isinstance(row, (list, tuple)):
                    # 元组或列表，使用列名索引
                    row_dict = {}
                    for i, value in enumerate(row):
                        if i < len(column_names):
                            col_name = column_names[i]
                            # 处理特殊类型
                            if isinstance(value, datetime):
                                row_dict[col_name] = value.isoformat()
                            elif value is None:
                                row_dict[col_name] = None
                            else:
                                row_dict[col_name] = str(value)
                else:
                    # 尝试直接转换为字典
                    try:
                        row_dict = dict(row)
                    except (TypeError, ValueError):
                        row_dict = {}

                # 统一处理日期时间类型
                for key, value in row_dict.items():
                    if isinstance(value, datetime):
                        row_dict[key] = value.isoformat()

                data.append(row_dict)

            # 更新产品的列数信息
            if product.column_count != len(columns):
                product.column_count = len(columns)
                db.session.commit()

            return {
                "product": product.to_dict(),
                "columns": columns,
                "data": data,
                "total_count": total_count,
                "preview_count": len(data),
            }

        except Exception as e:
            logger.error(f"获取数据预览失败: {str(e)}")
            raise

    @staticmethod
    def export_to_excel(
        product_id: int,
        limit: int = 200,
    ) -> tuple[io.BytesIO, str]:
        """
        导出数据产品数据为Excel文件

        Args:
            product_id: 数据产品ID
            limit: 导出数据条数，默认200

        Returns:
            (Excel文件字节流, 文件名)
        """
        try:
            # 延迟导入，避免启动时加载
            import pandas as pd

            product = DataProduct.query.get(product_id)
            if not product:
                raise ValueError(f"数据产品不存在: ID={product_id}")

            schema = product.target_schema or "public"
            table = product.target_table
            full_table_name = f"{schema}.{table}"

            # 检查表是否存在
            check_sql = text(
                """
                SELECT EXISTS (
                    SELECT FROM information_schema.tables
                    WHERE table_schema = :schema
                    AND table_name = :table
                )
                """
            )
            result = db.session.execute(
                check_sql, {"schema": schema, "table": table}
            ).scalar()

            if not result:
                raise ValueError(f"目标表 {full_table_name} 不存在")

            # 查询数据
            query_sql = text(f'SELECT * FROM "{schema}"."{table}" LIMIT :limit')
            result = db.session.execute(query_sql, {"limit": limit})

            # 获取列名
            column_names = list(result.keys())

            # 获取数据
            rows = result.fetchall()

            # 将 Row 对象转换为元组列表，以便 pandas 正确处理
            rows_data = [tuple(row) for row in rows]

            # 创建DataFrame
            # pandas DataFrame 构造函数接受列表和列名，类型检查器可能无法正确推断
            df = pd.DataFrame(rows_data, columns=column_names)  # type: ignore[arg-type]

            # 创建Excel文件
            output = io.BytesIO()
            # ExcelWriter 支持 BytesIO，类型检查器可能无法正确推断
            with pd.ExcelWriter(output, engine="openpyxl") as writer:  # type: ignore[arg-type]
                df.to_excel(writer, index=False, sheet_name="数据预览")

            output.seek(0)

            # 生成文件名
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"{product.product_name_en}_{timestamp}.xlsx"

            logger.info(f"导出Excel成功: product_id={product_id}, rows={len(rows)}")

            return output, filename

        except Exception as e:
            logger.error(f"导出Excel失败: {str(e)}")
            raise

    @staticmethod
    def mark_as_viewed(product_id: int) -> DataProduct | None:
        """
        标记数据产品为已查看

        Args:
            product_id: 数据产品ID

        Returns:
            更新后的数据产品对象
        """
        try:
            product = DataProduct.query.get(product_id)
            if not product:
                return None

            product.mark_as_viewed()
            db.session.commit()

            logger.info(f"标记数据产品为已查看: product_id={product_id}")
            return product

        except Exception as e:
            db.session.rollback()
            logger.error(f"标记已查看失败: {str(e)}")
            raise

    @staticmethod
    def register_data_product(
        product_name: str,
        product_name_en: str,
        target_table: str,
        target_schema: str = "public",
        description: str | None = None,
        source_dataflow_id: int | None = None,
        source_dataflow_name: str | None = None,
        created_by: str = "system",
    ) -> DataProduct:
        """
        注册新的数据产品

        Args:
            product_name: 数据产品名称（中文）
            product_name_en: 数据产品英文名
            target_table: 目标表名
            target_schema: 目标schema
            description: 描述
            source_dataflow_id: 关联的数据流ID
            source_dataflow_name: 数据流名称
            created_by: 创建人

        Returns:
            创建的数据产品对象
        """
        try:
            # 检查是否已存在
            existing = DataProduct.query.filter_by(
                target_schema=target_schema,
                target_table=target_table,
            ).first()

            if existing:
                # 更新现有记录
                existing.product_name = product_name
                existing.product_name_en = product_name_en
                existing.description = description
                existing.source_dataflow_id = source_dataflow_id
                existing.source_dataflow_name = source_dataflow_name
                existing.updated_at = datetime.utcnow()
                existing.last_updated_at = datetime.utcnow()
                db.session.commit()

                logger.info(
                    f"更新数据产品: {product_name} -> {target_schema}.{target_table}"
                )
                return existing

            # 创建新记录
            # SQLAlchemy 模型支持关键字参数初始化，类型检查器可能无法正确推断
            # pyright: ignore[reportCallIssue]
            product = DataProduct(
                product_name=product_name,  # type: ignore[arg-type]
                product_name_en=product_name_en,  # type: ignore[arg-type]
                target_table=target_table,  # type: ignore[arg-type]
                target_schema=target_schema,  # type: ignore[arg-type]
                description=description,  # type: ignore[arg-type]
                source_dataflow_id=source_dataflow_id,  # type: ignore[arg-type]
                source_dataflow_name=source_dataflow_name,  # type: ignore[arg-type]
                created_by=created_by,  # type: ignore[arg-type]
                last_updated_at=datetime.utcnow(),  # type: ignore[arg-type]
            )

            db.session.add(product)
            db.session.commit()

            logger.info(
                f"注册数据产品成功: {product_name} -> {target_schema}.{target_table}"
            )
            return product

        except Exception as e:
            db.session.rollback()
            logger.error(f"注册数据产品失败: {str(e)}")
            raise

    @staticmethod
    def update_data_stats(
        product_id: int,
        record_count: int | None = None,
        column_count: int | None = None,
    ) -> DataProduct | None:
        """
        更新数据产品的统计信息

        Args:
            product_id: 数据产品ID
            record_count: 记录数
            column_count: 列数

        Returns:
            更新后的数据产品对象
        """
        try:
            product = DataProduct.query.get(product_id)
            if not product:
                return None

            if record_count is not None:
                product.record_count = record_count
            if column_count is not None:
                product.column_count = column_count

            product.last_updated_at = datetime.utcnow()
            product.updated_at = datetime.utcnow()

            db.session.commit()

            logger.info(
                f"更新数据产品统计: product_id={product_id}, "
                f"record_count={record_count}, column_count={column_count}"
            )
            return product

        except Exception as e:
            db.session.rollback()
            logger.error(f"更新数据统计失败: {str(e)}")
            raise

    @staticmethod
    def refresh_product_stats(product_id: int) -> DataProduct | None:
        """
        刷新数据产品的统计信息（从目标表重新统计）

        Args:
            product_id: 数据产品ID

        Returns:
            更新后的数据产品对象
        """
        try:
            product = DataProduct.query.get(product_id)
            if not product:
                return None

            schema = product.target_schema or "public"
            table = product.target_table

            # 检查表是否存在
            check_sql = text(
                """
                SELECT EXISTS (
                    SELECT FROM information_schema.tables
                    WHERE table_schema = :schema
                    AND table_name = :table
                )
                """
            )
            exists = db.session.execute(
                check_sql, {"schema": schema, "table": table}
            ).scalar()

            if not exists:
                product.status = "error"
                product.updated_at = datetime.utcnow()
                db.session.commit()
                return product

            # 获取记录数
            count_sql = text(f'SELECT COUNT(*) FROM "{schema}"."{table}"')
            record_count = db.session.execute(count_sql).scalar() or 0

            # 获取列数
            columns_sql = text(
                """
                SELECT COUNT(*)
                FROM information_schema.columns
                WHERE table_schema = :schema AND table_name = :table
                """
            )
            column_count = (
                db.session.execute(
                    columns_sql, {"schema": schema, "table": table}
                ).scalar()
                or 0
            )

            # 更新统计信息
            product.record_count = record_count
            product.column_count = column_count
            product.last_updated_at = datetime.utcnow()
            product.updated_at = datetime.utcnow()
            product.status = "active"

            db.session.commit()

            logger.info(
                f"刷新数据产品统计: product_id={product_id}, "
                f"record_count={record_count}, column_count={column_count}"
            )
            return product

        except Exception as e:
            db.session.rollback()
            logger.error(f"刷新数据统计失败: {str(e)}")
            raise

    @staticmethod
    def delete_product(product_id: int) -> bool:
        """
        删除数据产品

        Args:
            product_id: 数据产品ID

        Returns:
            是否删除成功
        """
        try:
            product = DataProduct.query.get(product_id)
            if not product:
                return False

            db.session.delete(product)
            db.session.commit()

            logger.info(f"删除数据产品成功: product_id={product_id}")
            return True

        except Exception as e:
            db.session.rollback()
            logger.error(f"删除数据产品失败: {str(e)}")
            raise

    @staticmethod
    def get_data_lineage_visualization(
        product_id: int,
        sample_data: dict[str, Any],
    ) -> dict[str, Any]:
        """
        获取数据加工可视化血缘图谱

        从数据产品关联的目标 BusinessDomain 节点开始，逆向追溯数据生产链条：
        1. 根据 product_id 找到 DataProduct，确定目标 BusinessDomain
        2. 从目标 BusinessDomain 通过 OUTPUT 关系（反向）找到 DataFlow 节点
        3. 获取 DataFlow 的 script_requirement 属性作为数据流程定义
        4. 通过 INPUT 关系找到上游 BusinessDomain 节点
        5. 根据 sample_data 的键值在各节点中查找对应的数据
        6. 递归直到 BusinessDomain 没有被 DataFlow OUTPUT 指向为止

        Args:
            product_id: 数据产品ID
            sample_data: 前端传入的单条样例数据（JSON对象，key为字段名）

        Returns:
            包含完整血缘信息的字典：
            - nodes: 所有节点列表（BusinessDomain 和 DataFlow）
            - lines: 所有关系列表（INPUT 和 OUTPUT）
            - lineage_depth: 追溯深度
        """
        try:
            # 1. 获取数据产品信息
            product = DataProduct.query.get(product_id)
            if not product:
                raise ValueError(f"数据产品不存在: ID={product_id}")

            logger.info(
                f"开始血缘追溯: product_id={product_id}, "
                f"target_table={product.target_table}"
            )

            # 2. 找到目标 BusinessDomain
            target_bd_id = None

            with neo4j_driver.get_session() as session:
                # 方式1：通过 DataFlow 的 OUTPUT 关系找到目标 BusinessDomain
                if product.source_dataflow_id:
                    query = """
                    MATCH (df:DataFlow)-[:OUTPUT]->(bd:BusinessDomain)
                    WHERE id(df) = $dataflow_id
                    RETURN id(bd) as bd_id, bd.name_zh as name_zh, bd.name_en as name_en
                    LIMIT 1
                    """
                    result = session.run(
                        query, {"dataflow_id": product.source_dataflow_id}
                    ).single()
                    if result:
                        target_bd_id = result["bd_id"]
                        logger.info(
                            f"通过DataFlow找到目标BusinessDomain: "
                            f"{result['name_zh']} (ID: {target_bd_id})"
                        )

                # 方式2：通过表名匹配
                if not target_bd_id:
                    query = """
                    MATCH (bd:BusinessDomain)
                    WHERE bd.name_en = $table_name OR bd.name = $table_name
                    RETURN id(bd) as bd_id, bd.name_zh as name_zh, bd.name_en as name_en
                    LIMIT 1
                    """
                    result = session.run(
                        query, {"table_name": product.target_table}
                    ).single()
                    if result:
                        target_bd_id = result["bd_id"]
                        logger.info(
                            f"通过表名找到目标BusinessDomain: "
                            f"{result['name_zh']} (ID: {target_bd_id})"
                        )

                if not target_bd_id:
                    logger.warning(f"未找到数据产品关联的BusinessDomain: {product_id}")
                    return {
                        "nodes": [],
                        "lines": [],
                        "lineage_depth": 0,
                        "error": "未找到关联的业务领域节点",
                    }

                # 3. 递归追溯血缘并获取数据流程定义
                result = DataProductService._trace_production_chain(
                    session, target_bd_id, sample_data
                )

                logger.info(
                    f"血缘追溯完成: product_id={product_id}, "
                    f"nodes={len(result['nodes'])}, "
                    f"lines={len(result['lines'])}, "
                    f"depth={result['lineage_depth']}"
                )

                return result

        except Exception as e:
            logger.error(f"获取血缘可视化数据失败: {str(e)}")
            raise

    @staticmethod
    def _trace_production_chain(
        session: Any,
        target_bd_id: int,
        sample_data: dict[str, Any],
        max_depth: int = 10,
    ) -> dict[str, Any]:
        """
        递归追溯数据生产链条

        追溯逻辑（从目标节点向上游追溯）：
        1. 从当前 BusinessDomain 找到通过 OUTPUT 关系指向它的 DataFlow
        2. 获取 DataFlow 的 script_requirement 作为数据流程定义
        3. 从 DataFlow 找到通过 INPUT 关系连接的上游 BusinessDomain
        4. 根据 sample_data 的键值在各 BusinessDomain 中匹配数据
        5. 递归重复直到 BusinessDomain 没有被 DataFlow OUTPUT 指向

        Args:
            session: Neo4j会话
            target_bd_id: 目标 BusinessDomain 节点ID
            sample_data: 样例数据用于字段匹配
            max_depth: 最大追溯深度

        Returns:
            包含 nodes, lines, lineage_depth 的字典
        """
        nodes: list[dict[str, Any]] = []
        lines: list[dict[str, Any]] = []
        visited_bd: set[int] = set()
        visited_df: set[int] = set()

        def trace_upstream(bd_id: int, depth: int) -> int:
            """递归追溯上游生产链条"""
            if depth >= max_depth or bd_id in visited_bd:
                return depth

            visited_bd.add(bd_id)

            # 获取 BusinessDomain 节点信息和字段
            bd_query = """
            MATCH (bd:BusinessDomain)
            WHERE id(bd) = $bd_id
            OPTIONAL MATCH (bd)-[inc:INCLUDES]->(m:DataMeta)
            OPTIONAL MATCH (m)-[:LABEL]->(label:DataLabel)
            RETURN bd, labels(bd) as bd_labels,
                   collect(DISTINCT {
                       meta_id: id(m),
                       name_zh: coalesce(inc.alias_name_zh, m.name_zh),
                       name_en: coalesce(inc.alias_name_en, m.name_en),
                       data_type: m.data_type,
                       tags: collect(DISTINCT {id: id(label), name_zh: label.name_zh})
                   }) as fields
            """
            bd_result = session.run(bd_query, {"bd_id": bd_id}).single()
            if not bd_result:
                return depth

            bd_node = dict(bd_result["bd"])
            bd_labels = bd_result["bd_labels"]
            raw_fields = bd_result.get("fields", [])

            # 处理字段，过滤空值并匹配数据
            fields = [f for f in raw_fields if f.get("meta_id") is not None]
            for field in fields:
                field["tags"] = [t for t in field.get("tags", []) if t.get("id")]

            # 匹配 sample_data 到字段
            matched_data = {}
            for field in fields:
                name_zh = field.get("name_zh", "")
                name_en = field.get("name_en", "")
                if name_zh and name_zh in sample_data:
                    matched_data[name_zh] = sample_data[name_zh]
                elif name_en and name_en in sample_data:
                    matched_data[name_en] = sample_data[name_en]

            # 添加 BusinessDomain 节点
            nodes.append(
                {
                    "id": bd_id,
                    "node_type": "BusinessDomain",
                    "name_zh": bd_node.get("name_zh") or bd_node.get("name", ""),
                    "name_en": bd_node.get("name_en", ""),
                    "labels": bd_labels,
                    "depth": depth,
                    "is_target": depth == 0,
                    "is_source": "DataResource" in bd_labels,
                    "fields": fields,
                    "matched_data": matched_data,
                }
            )

            # 查找通过 OUTPUT 关系指向当前 BD 的 DataFlow
            df_query = """
            MATCH (df:DataFlow)-[:OUTPUT]->(bd:BusinessDomain)
            WHERE id(bd) = $bd_id
            RETURN df, id(df) as df_id, labels(df) as df_labels
            """
            df_results = session.run(df_query, {"bd_id": bd_id}).data()

            if not df_results:
                return depth  # 无上游，停止追溯

            max_depth_reached = depth

            for df_record in df_results:
                df_id = df_record["df_id"]
                if df_id in visited_df:
                    continue
                visited_df.add(df_id)

                df_node = dict(df_record["df"])

                # 添加 DataFlow 节点
                nodes.append(
                    {
                        "id": df_id,
                        "node_type": "DataFlow",
                        "name_zh": df_node.get("name_zh") or df_node.get("name", ""),
                        "name_en": df_node.get("name_en", ""),
                        "labels": df_record["df_labels"],
                        "depth": depth,
                        "script_requirement": df_node.get("script_requirement", ""),
                        "script_name": df_node.get("script_name", ""),
                        "script_type": df_node.get("script_type", ""),
                        "update_mode": df_node.get("update_mode", ""),
                    }
                )

                # 添加 OUTPUT 关系
                lines.append(
                    {
                        "from_id": df_id,
                        "to_id": bd_id,
                        "type": "OUTPUT",
                    }
                )

                # 查找上游 BusinessDomain
                input_query = """
                MATCH (source:BusinessDomain)-[:INPUT]->(df:DataFlow)
                WHERE id(df) = $df_id
                RETURN id(source) as source_id
                """
                input_results = session.run(input_query, {"df_id": df_id}).data()

                for input_record in input_results:
                    source_id = input_record["source_id"]

                    # 添加 INPUT 关系
                    lines.append(
                        {
                            "from_id": source_id,
                            "to_id": df_id,
                            "type": "INPUT",
                        }
                    )

                    # 递归追溯上游
                    reached = trace_upstream(source_id, depth + 1)
                    max_depth_reached = max(max_depth_reached, reached)

            return max_depth_reached

        actual_depth = trace_upstream(target_bd_id, 0)

        return {
            "nodes": nodes,
            "lines": lines,
            "lineage_depth": actual_depth,
        }


class DataOrderService:
    """数据订单服务类"""

    @staticmethod
    def _generate_order_no() -> str:
        """
        生成订单编号

        Returns:
            订单编号，格式：DO + 年月日 + 4位序号
        """
        today = datetime.now().strftime("%Y%m%d")
        prefix = f"DO{today}"

        # 查询今日最大序号
        last_order = (
            DataOrder.query.filter(DataOrder.order_no.like(f"{prefix}%"))
            .order_by(DataOrder.order_no.desc())
            .first()
        )

        if last_order:
            try:
                last_seq = int(last_order.order_no[-4:])
                new_seq = last_seq + 1
            except ValueError:
                new_seq = 1
        else:
            new_seq = 1

        return f"{prefix}{new_seq:04d}"

    @staticmethod
    def get_orders(
        page: int = 1,
        page_size: int = 20,
        search: str = "",
        status: str | None = None,
    ) -> dict[str, Any]:
        """
        获取数据订单列表

        Args:
            page: 页码
            page_size: 每页大小
            search: 搜索关键词
            status: 状态过滤

        Returns:
            包含数据订单列表和分页信息的字典
        """
        try:
            query = DataOrder.query

            # 搜索过滤
            if search:
                search_pattern = f"%{search}%"
                query = query.filter(
                    db.or_(
                        DataOrder.order_no.ilike(search_pattern),
                        DataOrder.title.ilike(search_pattern),
                        DataOrder.description.ilike(search_pattern),
                    )
                )

            # 状态过滤
            if status is not None:
                query = query.filter(DataOrder.status == status)  # pyright: ignore[reportArgumentType]

            # 计算总数
            total = query.count()

            # 分页查询
            orders = (
                query.order_by(DataOrder.created_at.desc())
                .offset((page - 1) * page_size)
                .limit(page_size)
                .all()
            )

            # 转换为字典列表
            order_list = [order.to_dict() for order in orders]

            return {
                "list": order_list,
                "pagination": {
                    "page": page,
                    "page_size": page_size,
                    "total": total,
                    "total_pages": (total + page_size - 1) // page_size,
                },
            }

        except Exception as e:
            logger.error(f"获取数据订单列表失败: {str(e)}")
            raise

    @staticmethod
    def get_order_by_id(order_id: int) -> DataOrder | None:
        """
        根据ID获取数据订单

        Args:
            order_id: 数据订单ID

        Returns:
            数据订单对象，不存在则返回None
        """
        return DataOrder.query.get(order_id)

    @staticmethod
    def create_order(
        title: str,
        description: str,
        created_by: str = "user",
    ) -> DataOrder:
        """
        创建数据订单

        Args:
            title: 订单标题
            description: 需求描述
            created_by: 创建人

        Returns:
            创建的数据订单对象
        """
        try:
            order_no = DataOrderService._generate_order_no()

            order = DataOrder(
                order_no=order_no,  # type: ignore[arg-type]
                title=title,  # type: ignore[arg-type]
                description=description,  # type: ignore[arg-type]
                status=DataOrder.STATUS_PENDING,  # type: ignore[arg-type]
                created_by=created_by,  # type: ignore[arg-type]
            )

            db.session.add(order)
            db.session.commit()

            logger.info(f"创建数据订单成功: order_no={order_no}")
            return order

        except Exception as e:
            db.session.rollback()
            logger.error(f"创建数据订单失败: {str(e)}")
            raise

    @staticmethod
    def extract_entities(description: str) -> dict[str, Any]:
        """
        使用 LLM 从描述中提取业务领域和数据字段

        Args:
            description: 需求描述

        Returns:
            提取结果，包含 business_domains, data_fields, purpose
        """
        try:
            client = OpenAI(
                api_key=current_app.config.get("LLM_API_KEY"),
                base_url=current_app.config.get("LLM_BASE_URL"),
            )

            model = current_app.config.get("LLM_MODEL_NAME")

            prompt = f"""分析以下数据需求描述，提取其中涉及的业务领域和数据字段。

需求描述：{description}

请严格按照以下JSON格式返回，不要添加任何解释或其他内容：
{{
    "business_domains": ["业务领域名称1", "业务领域名称2"],
    "data_fields": ["字段名称1", "字段名称2"],
    "purpose": "数据用途简述"
}}

注意：
1. business_domains 应该是可能存在的数据表或业务实体名称，如"人员信息"、"薪资数据"、"销售记录"等
2. data_fields 应该是具体的数据字段名称，如"姓名"、"年龄"、"薪资"、"销售额"等
3. purpose 简要描述数据的使用目的
"""

            completion = client.chat.completions.create(
                model=model,  # type: ignore[arg-type]
                messages=[
                    {
                        "role": "system",
                        "content": "你是一个专业的数据分析师，擅长从自然语言描述中提取数据需求。"
                        "请严格按照要求的JSON格式返回结果。",
                    },
                    {"role": "user", "content": prompt},
                ],
                temperature=0.1,
                max_tokens=1024,
            )

            response_text = (
                completion.choices[0].message.content.strip()  # type: ignore[union-attr]
            )

            # 尝试解析 JSON
            # 清理可能的 markdown 代码块标记
            if response_text.startswith("```"):
                lines = response_text.split("\n")
                # 移除首尾的代码块标记
                if lines[0].startswith("```"):
                    lines = lines[1:]
                if lines and lines[-1].strip() == "```":
                    lines = lines[:-1]
                response_text = "\n".join(lines)

            result = json.loads(response_text)

            logger.info(f"LLM 实体提取成功: {result}")
            return result

        except json.JSONDecodeError as e:
            logger.error(f"LLM 返回结果解析失败: {str(e)}, response: {response_text}")
            return {
                "business_domains": [],
                "data_fields": [],
                "purpose": "",
                "error": "解析失败",
            }
        except Exception as e:
            logger.error(f"LLM 实体提取失败: {str(e)}")
            return {
                "business_domains": [],
                "data_fields": [],
                "purpose": "",
                "error": str(e),
            }

    @staticmethod
    def find_matching_domains(domain_names: list[str]) -> list[dict[str, Any]]:
        """
        在 Neo4j 中查找匹配的 BusinessDomain 节点

        Args:
            domain_names: 业务领域名称列表

        Returns:
            匹配的 BusinessDomain 节点列表
        """
        try:
            with neo4j_driver.get_session() as session:
                # 使用模糊匹配查找 BusinessDomain
                cypher = """
                UNWIND $domain_names AS name
                MATCH (bd:BusinessDomain)
                WHERE bd.name_zh CONTAINS name OR name CONTAINS bd.name_zh
                   OR bd.name_en CONTAINS name OR name CONTAINS bd.name_en
                RETURN DISTINCT id(bd) as id, bd.name_zh as name_zh,
                       bd.name_en as name_en, bd.describe as describe
                """
                result = session.run(cypher, {"domain_names": domain_names})

                domains = []
                for record in result:
                    domains.append(
                        {
                            "id": record["id"],
                            "name_zh": record["name_zh"],
                            "name_en": record["name_en"],
                            "describe": record["describe"],
                        }
                    )

                logger.info(f"找到 {len(domains)} 个匹配的 BusinessDomain")
                return domains

        except Exception as e:
            logger.error(f"查找匹配的 BusinessDomain 失败: {str(e)}")
            return []

    @staticmethod
    def find_matching_fields(field_names: list[str]) -> list[dict[str, Any]]:
        """
        在 Neo4j 中查找匹配的 DataMeta 节点

        Args:
            field_names: 字段名称列表

        Returns:
            匹配的 DataMeta 节点列表
        """
        try:
            with neo4j_driver.get_session() as session:
                # 使用模糊匹配查找 DataMeta
                cypher = """
                UNWIND $field_names AS name
                MATCH (m:DataMeta)
                WHERE m.name_zh CONTAINS name OR name CONTAINS m.name_zh
                   OR m.name_en CONTAINS name OR name CONTAINS m.name_en
                RETURN DISTINCT id(m) as id, m.name_zh as name_zh,
                       m.name_en as name_en, m.data_type as data_type
                """
                result = session.run(cypher, {"field_names": field_names})

                fields = []
                for record in result:
                    fields.append(
                        {
                            "id": record["id"],
                            "name_zh": record["name_zh"],
                            "name_en": record["name_en"],
                            "data_type": record["data_type"],
                        }
                    )

                logger.info(f"找到 {len(fields)} 个匹配的 DataMeta")
                return fields

        except Exception as e:
            logger.error(f"查找匹配的 DataMeta 失败: {str(e)}")
            return []

    @staticmethod
    def analyze_graph_connection(
        domain_ids: list[int],
    ) -> dict[str, Any]:
        """
        分析多个 BusinessDomain 之间的连通性（通过共同的 DataMeta 字段）

        Args:
            domain_ids: BusinessDomain 节点 ID 列表

        Returns:
            连通性分析结果
        """
        try:
            if len(domain_ids) < 2:
                return {
                    "can_connect": len(domain_ids) == 1,
                    "reason": "至少需要两个业务领域才能分析连通性"
                    if len(domain_ids) < 1
                    else "单个业务领域无需连接",
                    "common_fields": [],
                    "connection_pairs": [],
                }

            with neo4j_driver.get_session() as session:
                # 查找多个 BusinessDomain 之间的共同 DataMeta 字段
                cypher = """
                MATCH (bd1:BusinessDomain)-[:INCLUDES]->(m:DataMeta)<-[:INCLUDES]-(bd2:BusinessDomain)
                WHERE id(bd1) IN $domain_ids AND id(bd2) IN $domain_ids
                AND id(bd1) < id(bd2)
                RETURN id(bd1) as bd1_id, bd1.name_zh as bd1_name,
                       id(bd2) as bd2_id, bd2.name_zh as bd2_name,
                       collect({
                           id: id(m),
                           name_zh: m.name_zh,
                           name_en: m.name_en
                       }) as common_fields
                """
                result = session.run(cypher, {"domain_ids": domain_ids})

                connection_pairs = []
                all_common_fields = []

                for record in result:
                    pair = {
                        "domain1": {
                            "id": record["bd1_id"],
                            "name": record["bd1_name"],
                        },
                        "domain2": {
                            "id": record["bd2_id"],
                            "name": record["bd2_name"],
                        },
                        "common_fields": record["common_fields"],
                    }
                    connection_pairs.append(pair)
                    all_common_fields.extend(record["common_fields"])

                # 去重共同字段
                seen_ids = set()
                unique_fields = []
                for field in all_common_fields:
                    if field["id"] not in seen_ids:
                        seen_ids.add(field["id"])
                        unique_fields.append(field)

                can_connect = len(connection_pairs) > 0

                # 检查是否所有领域都有连接
                connected_domains = set()
                for pair in connection_pairs:
                    connected_domains.add(pair["domain1"]["id"])
                    connected_domains.add(pair["domain2"]["id"])

                all_connected = len(connected_domains) == len(domain_ids)

                analysis_result = {
                    "can_connect": can_connect,
                    "all_domains_connected": all_connected,
                    "connected_domain_count": len(connected_domains),
                    "total_domain_count": len(domain_ids),
                    "common_fields": unique_fields,
                    "connection_pairs": connection_pairs,
                    "reason": "找到可用于 JOIN 的共同字段"
                    if can_connect
                    else "未找到可用于 JOIN 的共同字段",
                }

                logger.info(f"图谱连通性分析完成: can_connect={can_connect}")
                return analysis_result

        except Exception as e:
            logger.error(f"图谱连通性分析失败: {str(e)}")
            return {
                "can_connect": False,
                "error": str(e),
                "common_fields": [],
                "connection_pairs": [],
            }

    @staticmethod
    def analyze_order(order_id: int) -> DataOrder | None:
        """
        分析数据订单：提取实体并检测图谱连通性

        Args:
            order_id: 订单ID

        Returns:
            更新后的订单对象
        """
        try:
            order = DataOrder.query.get(order_id)
            if not order:
                return None

            # 更新状态为分析中
            order.update_status(DataOrder.STATUS_ANALYZING)
            db.session.commit()

            # 1. LLM 提取实体
            extraction_result = DataOrderService.extract_entities(order.description)

            if extraction_result.get("error"):
                # 提取失败，标记为待补充
                order.update_status(DataOrder.STATUS_NEED_SUPPLEMENT)
                order.set_extraction_result(
                    domains=extraction_result.get("business_domains"),
                    fields=extraction_result.get("data_fields"),
                    purpose=extraction_result.get("purpose"),
                )
                db.session.commit()
                return order

            domains = extraction_result.get("business_domains", [])
            fields = extraction_result.get("data_fields", [])
            purpose = extraction_result.get("purpose", "")

            order.set_extraction_result(
                domains=domains,
                fields=fields,
                purpose=purpose,
            )

            # 2. 在图谱中查找匹配的节点
            matched_domains = DataOrderService.find_matching_domains(domains)
            matched_fields = DataOrderService.find_matching_fields(fields)

            if not matched_domains:
                # 没有找到匹配的业务领域，需要人工处理
                order.set_graph_analysis(
                    analysis={
                        "matched_domains": [],
                        "matched_fields": matched_fields,
                        "reason": "未找到匹配的业务领域",
                    },
                    can_connect=False,
                )
                order.update_status(DataOrder.STATUS_MANUAL_REVIEW)
                db.session.commit()
                return order

            # 3. 分析连通性
            domain_ids = [d["id"] for d in matched_domains]
            connection_result = DataOrderService.analyze_graph_connection(domain_ids)

            # 保存分析结果
            analysis = {
                "matched_domains": matched_domains,
                "matched_fields": matched_fields,
                "connection_analysis": connection_result,
            }

            can_connect = connection_result.get("can_connect", False)
            connection_path = None

            if can_connect:
                connection_path = {
                    "domains": [d["name_zh"] for d in matched_domains],
                    "join_fields": [
                        f["name_zh"] for f in connection_result.get("common_fields", [])
                    ],
                    "pairs": connection_result.get("connection_pairs", []),
                }

            order.set_graph_analysis(
                analysis=analysis,
                can_connect=can_connect,
                connection_path=connection_path,
            )

            # 根据连通性结果更新状态
            if can_connect:
                # 可连通，自动进入加工状态
                order.update_status(DataOrder.STATUS_PROCESSING)
            else:
                # 不可连通，需要人工处理
                order.update_status(DataOrder.STATUS_MANUAL_REVIEW)

            db.session.commit()
            logger.info(f"订单分析完成: order_id={order_id}, can_connect={can_connect}")
            return order

        except Exception as e:
            db.session.rollback()
            logger.error(f"分析数据订单失败: {str(e)}")
            raise

    @staticmethod
    def approve_order(
        order_id: int,
        processed_by: str = "admin",
    ) -> DataOrder | None:
        """
        审批通过订单

        Args:
            order_id: 订单ID
            processed_by: 处理人

        Returns:
            更新后的订单对象
        """
        try:
            order = DataOrder.query.get(order_id)
            if not order:
                return None

            if order.status != DataOrder.STATUS_MANUAL_REVIEW:
                raise ValueError(f"订单状态不允许审批: {order.status}")

            order.update_status(DataOrder.STATUS_PROCESSING, processed_by)
            db.session.commit()

            logger.info(
                f"订单审批通过: order_id={order_id}, processed_by={processed_by}"
            )
            return order

        except Exception as e:
            db.session.rollback()
            logger.error(f"审批订单失败: {str(e)}")
            raise

    @staticmethod
    def reject_order(
        order_id: int,
        reason: str,
        processed_by: str = "admin",
    ) -> DataOrder | None:
        """
        驳回订单

        Args:
            order_id: 订单ID
            reason: 驳回原因
            processed_by: 处理人

        Returns:
            更新后的订单对象
        """
        try:
            order = DataOrder.query.get(order_id)
            if not order:
                return None

            order.reject(reason, processed_by)
            db.session.commit()

            logger.info(
                f"订单已驳回: order_id={order_id}, reason={reason}, "
                f"processed_by={processed_by}"
            )
            return order

        except Exception as e:
            db.session.rollback()
            logger.error(f"驳回订单失败: {str(e)}")
            raise

    @staticmethod
    def complete_order(
        order_id: int,
        product_id: int | None = None,
        dataflow_id: int | None = None,
        processed_by: str = "system",
    ) -> DataOrder | None:
        """
        完成订单

        Args:
            order_id: 订单ID
            product_id: 生成的数据产品ID
            dataflow_id: 生成的数据流ID
            processed_by: 处理人

        Returns:
            更新后的订单对象
        """
        try:
            order = DataOrder.query.get(order_id)
            if not order:
                return None

            order.set_result(product_id, dataflow_id)
            order.update_status(DataOrder.STATUS_COMPLETED, processed_by)
            db.session.commit()

            logger.info(
                f"订单已完成: order_id={order_id}, product_id={product_id}, "
                f"dataflow_id={dataflow_id}"
            )
            return order

        except Exception as e:
            db.session.rollback()
            logger.error(f"完成订单失败: {str(e)}")
            raise

    @staticmethod
    def delete_order(order_id: int) -> bool:
        """
        删除数据订单

        Args:
            order_id: 数据订单ID

        Returns:
            是否删除成功
        """
        try:
            order = DataOrder.query.get(order_id)
            if not order:
                return False

            db.session.delete(order)
            db.session.commit()

            logger.info(f"删除数据订单成功: order_id={order_id}")
            return True

        except Exception as e:
            db.session.rollback()
            logger.error(f"删除数据订单失败: {str(e)}")
            raise
