2_qdrant_demo.py:

# qdrant_demo_fixed.py
import json
from dataclasses import dataclass
from typing import Optional, List, Dict, Any
from qdrant_client import QdrantClient
from qdrant_client.http import models
from qdrant_client.http.exceptions import UnexpectedResponse
import uuid
import os


# ========== 数据模型定义 ==========
@dataclass
class ColumnConfig:
    name: str
    role: str
    description: str
    alias: List[str]
    sync: bool

    def to_dict(self):
        return {
            "name": self.name,
            "role": self.role,
            "description": self.description,
            "alias": self.alias,
            "sync": self.sync
        }

    @classmethod
    def from_dict(cls, data):
        return cls(
            name=data["name"],
            role=data["role"],
            description=data["description"],
            alias=data.get("alias", []),
            sync=data.get("sync", False)
        )


@dataclass
class TableConfig:
    name: str
    role: str
    description: str
    columns: List[ColumnConfig]

    def to_dict(self):
        return {
            "name": self.name,
            "role": self.role,
            "description": self.description,
            "columns": [col.to_dict() for col in self.columns]
        }

    @classmethod
    def from_dict(cls, data):
        return cls(
            name=data["name"],
            role=data["role"],
            description=data["description"],
            columns=[ColumnConfig.from_dict(col) for col in data.get("columns", [])]
        )


@dataclass
class MetricConfig:
    name: str
    description: str
    relevant_columns: List[str]
    alias: List[str]

    def to_dict(self):
        return {
            "name": self.name,
            "description": self.description,
            "relevant_columns": self.relevant_columns,
            "alias": self.alias
        }

    @classmethod
    def from_dict(cls, data):
        return cls(
            name=data["name"],
            description=data["description"],
            relevant_columns=data.get("relevant_columns", []),
            alias=data.get("alias", [])
        )


@dataclass
class MetaConfig:
    tables: Optional[List[TableConfig]] = None
    metrics: Optional[List[MetricConfig]] = None

    def to_dict(self):
        result = {}
        if self.tables:
            result["tables"] = [table.to_dict() for table in self.tables]
        if self.metrics:
            result["metrics"] = [metric.to_dict() for metric in self.metrics]
        return result

    @classmethod
    def from_dict(cls, data):
        tables = None
        metrics = None
        if "tables" in data:
            tables = [TableConfig.from_dict(table) for table in data["tables"]]
        if "metrics" in data:
            metrics = [MetricConfig.from_dict(metric) for metric in data["metrics"]]
        return cls(tables=tables, metrics=metrics)


from data_agent.v0.day04_embed.emb_cpu_2 import EmbeddingClientManager

embedding_client_manager = EmbeddingClientManager()

embedding_client_manager.init()

query = embedding_client_manager.embed_query("hello world")

# 这里的EmbeddingClientManager  可以模拟 一个 也可以


# ========== Qdrant 连接配置 ==========
class QdrantConfig:
    """Qdrant配置管理"""

    @staticmethod
    def get_config():
        """获取Qdrant连接配置"""
        return {
            'host': os.getenv('QDRANT_HOST', 'localhost'),
            'port': int(os.getenv('QDRANT_PORT', 6333)),
            'api_key': os.getenv('QDRANT_API_KEY', None),  # 如果有API密钥
            'timeout': 30,
            'check_compatibility': False  # 忽略版本兼容性警告
        }

    @staticmethod
    def test_connection():
        """测试Qdrant连接"""
        config = QdrantConfig.get_config()
        try:
            client = QdrantClient(**config)
            client.get_collections()
            print("✅ Qdrant连接成功!")
            return True
        except UnexpectedResponse as e:
            print(f"❌ Qdrant连接失败: {e}")
            print("\n💡 解决方法:")
            print("   1. 如果Qdrant需要API密钥，设置环境变量:")
            print("      export QDRANT_API_KEY=your-api-key")
            print("   2. 或使用无认证的Qdrant:")
            print("      docker run -d --name qdrant-meta -p 6333:6333 qdrant/qdrant")
            return False
        except Exception as e:
            print(f"❌ Qdrant连接失败: {e}")
            return False


# ========== Qdrant 仓储实现 ==========
class QdrantMetaRepository:
    def __init__(self, host: str = "localhost", port: int = 6333,
                 collection_name: str = "demo_meta_configs",
                 api_key: str = "happy123",
                 check_compatibility: bool = False):

        # 构建客户端配置
        client_config = {
            'host': host,
            'port': port,
            'timeout': 30,
            'check_compatibility': check_compatibility,
            # TODO ?
             'https': False  # ✅ 强制使用 HTTP 协议
        }

        # 如果有API密钥，添加到配置中
        if api_key:
            client_config['api_key'] = api_key

        try:
            self.client = QdrantClient(**client_config)
            print(f"✅ Qdrant客户端初始化成功 (host={host}, port={port})")
        except Exception as e:
            print(f"❌ Qdrant客户端初始化失败: {e}")
            raise

        self.collection_name = collection_name
        self._init_collection()

    def _init_collection(self):
        """初始化集合"""
        try:
            # 检查集合是否存在
            collections = self.client.get_collections().collections
            collection_exists = any(c.name == self.collection_name for c in collections)

            if not collection_exists:
                # 创建集合
                self.client.create_collection(
                    collection_name=self.collection_name,
                    vectors_config=models.VectorParams(
                        # size=384,  # 使用all-MiniLM-L6-v2的维度
                        size=1024,  # ✅ 修改为 1024，与你的模型输出保持一致
                        distance=models.Distance.COSINE
                    )
                )
                print(f"✅ Qdrant集合 '{self.collection_name}' 创建成功")
            else:
                print(f"✅ Qdrant集合 '{self.collection_name}' 已存在")
        except UnexpectedResponse as e:
            print(f"❌ 集合操作失败: {e}")
            print("\n💡 可能的原因:")
            print("   1. Qdrant服务器未启动")
            print("   2. API密钥错误或未提供")
            print("   3. 端口或主机配置错误")
            raise
        except Exception as e:
            print(f"❌ 集合操作失败: {e}")
            raise

    def _generate_id(self) -> str:
        return str(uuid.uuid4())

    def _entity_to_vector(self, entity: MetaConfig) -> List[float]:
        """将实体转换为向量"""
        try:
            # 将配置转换为文本
            text = json.dumps(entity.to_dict(), ensure_ascii=False)
            # vector = embedding_client.embed_query(text)
            vector = embedding_client_manager.embed_query(text)
            return vector
        except Exception as e:
            print(f"⚠️ 向量生成失败: {e}")
            # 使用随机向量
            import random
            return [random.random() for _ in range(384)]

    def _prepare_payload(self, entity: MetaConfig) -> Dict[str, Any]:
        """准备payload"""
        return {
            "config_data": json.dumps(entity.to_dict(), ensure_ascii=False),
            "table_names": [t.name for t in entity.tables] if entity.tables else [],
            "metric_names": [m.name for m in entity.metrics] if entity.metrics else [],
            "has_tables": bool(entity.tables),
            "has_metrics": bool(entity.metrics)
        }

    # ========== CRUD 操作 ==========
    def create(self, entity: MetaConfig) -> str:
        """创建配置，返回ID"""
        point_id = self._generate_id()
        vector = self._entity_to_vector(entity)
        payload = self._prepare_payload(entity)

        try:
            self.client.upsert(
                collection_name=self.collection_name,
                points=[
                    models.PointStruct(
                        id=point_id,
                        vector=vector,
                        payload=payload
                    )
                ]
            )
            print(f"✅ Qdrant创建成功，ID: {point_id}")
            return point_id
        except Exception as e:
            print(f"❌ Qdrant创建失败: {e}")
            raise

    def get_by_id(self, id: str) -> Optional[MetaConfig]:
        """根据ID获取配置"""
        try:
            result = self.client.retrieve(
                collection_name=self.collection_name,
                ids=[id]
            )

            if result and len(result) > 0:
                payload = result[0].payload
                if payload and "config_data" in payload:
                    data = json.loads(payload["config_data"])
                    return MetaConfig.from_dict(data)
        except Exception as e:
            print(f"❌ 查询失败: {e}")
        return None

    def get_all(self, limit: int = 100) -> List[MetaConfig]:
        """获取所有配置"""
        try:
            results = self.client.scroll(
                collection_name=self.collection_name,
                limit=limit
            )

            configs = []
            for point in results[0]:
                if point.payload and "config_data" in point.payload:
                    data = json.loads(point.payload["config_data"])
                    configs.append(MetaConfig.from_dict(data))
            return configs
        except Exception as e:
            print(f"❌ 获取所有配置失败: {e}")
            return []

    def update(self, id: str, entity: MetaConfig) -> bool:
        """更新配置"""
        try:
            # 检查是否存在
            existing = self.get_by_id(id)
            if not existing:
                print(f"❌ ID {id} 不存在")
                return False

            vector = self._entity_to_vector(entity)
            payload = self._prepare_payload(entity)

            self.client.upsert(
                collection_name=self.collection_name,
                points=[
                    models.PointStruct(
                        id=id,
                        vector=vector,
                        payload=payload
                    )
                ]
            )
            print(f"✅ Qdrant更新成功，ID: {id}")
            return True
        except Exception as e:
            print(f"❌ Qdrant更新失败: {e}")
            return False

    def delete(self, id: str) -> bool:
        """删除配置"""
        try:
            result = self.client.delete(
                collection_name=self.collection_name,
                points_selector=models.PointIdsList(points=[id])
            )
            success = result.status == "completed"
            if success:
                print(f"✅ Qdrant删除成功，ID: {id}")
            else:
                print(f"❌ Qdrant删除失败")
            return success
        except Exception as e:
            print(f"❌ Qdrant删除失败: {e}")
            return False

    def search(self, query: str, limit: int = 10, score_threshold: float = 0.5) -> List[tuple]:
        """向量相似度搜索，返回 (配置, 相似度分数)"""
        try:
            # 生成查询向量
            query_vector = embedding_client_manager.embed_query(query)

            results = self.client.search(
                collection_name=self.collection_name,
                query_vector=query_vector,
                limit=limit,
                score_threshold=score_threshold
            )

            configs = []
            for scored_point in results:
                if scored_point.payload and "config_data" in scored_point.payload:
                    data = json.loads(scored_point.payload["config_data"])
                    config = MetaConfig.from_dict(data)
                    configs.append((config, scored_point.score))
            return configs
        except Exception as e:
            print(f"❌ 向量搜索失败: {e}")
            return []

    def search_by_keyword(self, query: str) -> List[MetaConfig]:
        """关键词搜索"""
        try:
            scroll_filter = models.Filter(
                should=[
                    models.FieldCondition(
                        key="table_names",
                        match=models.MatchText(text=query)
                    ),
                    models.FieldCondition(
                        key="metric_names",
                        match=models.MatchText(text=query)
                    )
                ]
            )

            results = self.client.scroll(
                collection_name=self.collection_name,
                scroll_filter=scroll_filter,
                limit=100
            )

            configs = []
            for point in results[0]:
                if point.payload and "config_data" in point.payload:
                    data = json.loads(point.payload["config_data"])
                    configs.append(MetaConfig.from_dict(data))
            return configs
        except Exception as e:
            print(f"❌ 关键词搜索失败: {e}")
            return []


# ========== 演示函数 ==========
def qdrant_demo():
    print("=" * 60)
    print("🔷 Qdrant 元数据管理演示")
    print("=" * 60)

    # 1. 检查并启动Qdrant
    print("\n📦 检查Qdrant服务...")

    try:
        repo = QdrantMetaRepository(
            host="localhost",
            port=6333,
            collection_name="demo_meta_configs",
            api_key="happy123",  # 无认证
            check_compatibility=False  # 忽略版本检查
        )
    except Exception as e:
        print(f"❌ 初始化失败: {e}")
        return

    # 4. 准备测试数据
    print("\n📦 准备测试数据...")

    columns = [
        ColumnConfig(
            name="user_id",
            role="primary_key",
            description="用户唯一标识",
            alias=["用户ID", "UID"],
            sync=False
        ),
        ColumnConfig(
            name="user_name",
            role="dimension",
            description="用户姓名",
            alias=["姓名", "用户名"],
            sync=True
        ),
        ColumnConfig(
            name="user_level",
            role="dimension",
            description="用户等级，如VIP、普通等",
            alias=["等级", "用户等级"],
            sync=True
        )
    ]

    tables = [
        TableConfig(
            name="dim_user",
            role="dim",
            description="用户维度表，存储用户基本信息",
            columns=columns
        )
    ]

    metrics = [
        MetricConfig(
            name="用户活跃度",
            description="用户在一定时间内的活跃程度评分",
            relevant_columns=["fact_user_activity.login_count", "fact_user_activity.action_count"],
            alias=["活跃度", "用户活跃分"]
        )
    ]

    meta_config = MetaConfig(tables=tables, metrics=metrics)

    # 5. 创建
    print("\n📝 创建配置...")
    try:
        config_id = repo.create(meta_config)
    except Exception as e:
        print(f"❌ 创建失败: {e}")
        return

    # 6. 查询
    print("\n🔍 根据ID查询...")
    retrieved = repo.get_by_id(config_id)
    if retrieved:
        print(f"  查询成功!")
        print(f"  表数量: {len(retrieved.tables) if retrieved.tables else 0}")
        print(f"  指标数量: {len(retrieved.metrics) if retrieved.metrics else 0}")
        if retrieved.tables:
            print(f"  表名: {[t.name for t in retrieved.tables]}")
        if retrieved.metrics:
            print(f"  指标名: {[m.name for m in retrieved.metrics]}")

    # 7. 向量搜索
    print("\n🔍 向量搜索 '用户相关配置'...")
    search_results = repo.search("用户相关配置", limit=3)
    if search_results:
        for config, score in search_results:
            table_names = [t.name for t in config.tables] if config.tables else []
            metric_names = [m.name for m in config.metrics] if config.metrics else []
            print(f"  相似度: {score:.4f}, 表: {table_names}, 指标: {metric_names}")
    else:
        print("  没有找到结果")

    # 8. 关键词搜索
    print("\n🔍 关键词搜索 'dim_user'...")
    keyword_results = repo.search_by_keyword("dim_user")
    print(f"  找到 {len(keyword_results)} 个配置")

    # 9. 更新
    print("\n📝 更新配置...")
    new_columns = columns + [
        ColumnConfig(
            name="email",
            role="dimension",
            description="用户邮箱",
            alias=["邮箱", "电子邮件"],
            sync=True
        )
    ]
    new_tables = [
        TableConfig(
            name="dim_user",
            role="dim",
            description="用户维度表，存储用户基本信息和联系方式",
            columns=new_columns
        )
    ]
    updated_config = MetaConfig(tables=new_tables, metrics=metrics)
    if repo.update(config_id, updated_config):
        print("  更新成功!")

    # 10. 获取所有
    print("\n📋 获取所有配置...")
    all_configs = repo.get_all()
    print(f"  共有 {len(all_configs)} 个配置")

    # 11. 删除
    print("\n🗑️ 删除配置...")
    repo.delete(config_id)
    print("  删除成功!")

    print("\n" + "=" * 60)
    print("✅ Qdrant演示完成!")
    print("=" * 60)


if __name__ == "__main__":
    qdrant_demo()

'''
Qdrant 自带了一个非常直观的 Web 管理界面，你可以直接在浏览器里访问，查看集合、数据和执行搜索。
🌐 访问 Qdrant Web UI
如果你的 Qdrant 服务正在本地运行，通常可以通过以下地址访问：
http://localhost:6333/dashboard
打开这个链接，你应该能看到一个图形化界面。
💻 在界面中查看数据
进入 Dashboard 后，你可以这样操作：
找到集合：在页面中找到并点击你的集合名称，也就是 demo_meta_configs。
查看数据点：进入集合详情页后，点击 "Points" 标签页。
浏览内容：这里会列出所有存入的数据点（Points）。你可以点击某个具体的 ID，查看它的：
Payload: 这里存储了你代码中的 MetaConfig 数据，比如表名、指标名等，是 JSON 格式。
Vector: 这里就是那个 1024 维的向量，通常是一串数字。
通过这个界面，你可以很方便地验证数据是否成功写入，以及检查数据的具体内容。
'''