from pathlib import Path

from omegaconf import OmegaConf

config_file = Path(__file__).parents[2] / "conf" / "app_config.yaml"

conf = OmegaConf.load(config_file)

from pathlib import Path

from omegaconf import OmegaConf

@dataclass

class Address:

    street: str

    city: str

    state: str

    zip: str

@dataclass

class AppConfig:

    name: str

    age: int

    height: float

    address: Address

config_file = Path(__file__).parents[2] / "conf" / "app_config.yaml"

content = OmegaConf.load(config_file)

struct = OmegaConf.structured(AppConfig)

app_conf:AppConfig = OmegaConf.to_object(OmegaConf.merge(struct, content))

print(app_conf.address.street)

from dataclasses import dataclass

from pathlib import Path

from omegaconf import OmegaConf

@dataclass

class File:

    enable: bool

    level: str

    path: str

    rotation: str

    retention: str

@dataclass

class Console:

    enable: bool

    level: str

@dataclass

class LoggingConfig:

    file: File

    console: Console

# 数据库配置

@dataclass

class DBConfig:

    host: str

    port: int

    user: str

    password: str

    database: str

@dataclass

class QdrantConfig:

    host: str

    port: int

    embedding_size: int

@dataclass

class EmbeddingConfig:

    host: str

    port: int

    model: str

@dataclass

class ESConfig:

    host: str

    port: int

    index_name: str

@dataclass

class LLMConfig:

    model_name: str

    api_key: str

    base_url: str

@dataclass

class AppConfig:

    logging: LoggingConfig

    db_meta: DBConfig

    db_dw: DBConfig

    qdrant: QdrantConfig

    embedding: EmbeddingConfig

    es: ESConfig

    llm: LLMConfig

config_file = Path(__file__).parents[2] / 'conf' / 'app_config.yaml'

context = OmegaConf.load(config_file)

schema = OmegaConf.structured(AppConfig)

app_config: AppConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

from qdrant_client import QdrantClient

from qdrant_client.models import Distance, VectorParams

from qdrant_client.models import PointStruct

from app.conf.app_config import QdrantConfig, app_config

class QdrantClientManager:

    def __init__(self, config: QdrantConfig):

        self.client: QdrantClient | None = None

        self.config: QdrantConfig = config

    def _get_url(self):

        return f"http://{self.config.host}:{self.config.port}"

    def init(self):

        self.client = QdrantClient(url=self._get_url())

    def close(self):

        self.client.close()

qdrant_client_manager = QdrantClientManager(app_config.qdrant)

if __name__ == '__main__':

    qdrant_client_manager.init()

    client = qdrant_client_manager.client

    client.create_collection(

        collection_name="test_collection",

        vectors_config=VectorParams(size=4, distance=Distance.COSINE),

    )

    operation_info = client.upsert(

        collection_name="test_collection",

        wait=True,

        points=[

            PointStruct(id=1, vector=[0.05, 0.61, 0.76, 0.74], payload={"city": "Berlin"}),

            PointStruct(id=2, vector=[0.19, 0.81, 0.75, 0.11], payload={"city": "London"}),

            PointStruct(id=3, vector=[0.36, 0.55, 0.47, 0.94], payload={"city": "Moscow"}),

            PointStruct(id=4, vector=[0.18, 0.01, 0.85, 0.80], payload={"city": "New York"}),

            PointStruct(id=5, vector=[0.24, 0.18, 0.22, 0.44], payload={"city": "Beijing"}),

            PointStruct(id=6, vector=[0.35, 0.08, 0.11, 0.44], payload={"city": "Mumbai"}),

        ],

    )

    search_result = client.query_points(

        collection_name="test_collection",

        query=[0.2, 0.1, 0.9, 0.7],

        with_payload=False,

        limit=3

    ).points

    print(search_result)

import asyncio

from qdrant_client import AsyncQdrantClient

from qdrant_client.models import Distance, VectorParams

from qdrant_client.models import PointStruct

from app.conf.app_config import QdrantConfig, app_config

class QdrantClientManager:

    def __init__(self, config: QdrantConfig):

        self.client: AsyncQdrantClient | None = None

        self.config: QdrantConfig = config

    def _get_url(self):

        return f"http://{self.config.host}:{self.config.port}"

    def init(self):

        self.client = AsyncQdrantClient(url=self._get_url())

    async def close(self):

        await self.client.close()

qdrant_client_manager = QdrantClientManager(app_config.qdrant)

if __name__ == '__main__':

    qdrant_client_manager.init()

    client = qdrant_client_manager.client

    async def test():

        await client.create_collection(

            collection_name="test_collection_async",

            vectors_config=VectorParams(size=4, distance=Distance.COSINE),

        )

        await client.upsert(

            collection_name="test_collection_async",

            wait=True,

            points=[

                PointStruct(id=1, vector=[0.05, 0.61, 0.76, 0.74], payload={"city": "Berlin"}),

                PointStruct(id=2, vector=[0.19, 0.81, 0.75, 0.11], payload={"city": "London"}),

                PointStruct(id=3, vector=[0.36, 0.55, 0.47, 0.94], payload={"city": "Moscow"}),

                PointStruct(id=4, vector=[0.18, 0.01, 0.85, 0.80], payload={"city": "New York"}),

                PointStruct(id=5, vector=[0.24, 0.18, 0.22, 0.44], payload={"city": "Beijing"}),

                PointStruct(id=6, vector=[0.35, 0.08, 0.11, 0.44], payload={"city": "Mumbai"}),

            ],

        )

        search_result = await client.query_points(

            collection_name="test_collection_async",

            query=[0.2, 0.1, 0.9, 0.7],

            with_payload=False,

            limit=3

        )

        print(search_result.points)

    asyncio.run(test())

from elasticsearch import AsyncElasticsearch

from app.conf.app_config import ESConfig, app_config

class ESClientManager:

    def __init__(self, config: ESConfig):

        self.config: ESConfig = config

        self.client: AsyncElasticsearch | None = None

    def _get_url(self):

        return f"http://{self.config.host}:{self.config.port}"

    def init(self):

        self.client = AsyncElasticsearch(

            hosts=[self._get_url()]

        )

    async def close(self):

        if self.client:

            await self.client.close()

es_client_manager = ESClientManager(app_config.es)

if __name__ == '__main__':

    es_client_manager.init()

    client = es_client_manager.client

    async def test():

        # 创建索引

        await client.indices.create(index="test_index", ignore=400)

        await client.index(

            index="books",

            document={

                "title": "The Great Gatsby",

                "author": "F. Scott Fitzgerald",

                "date": "1925-04-10",

                "summary": "A novel set in the Jazz Age that tells the story of Jay Gatsby's unrequited love for Daisy Buchanan.",

            }

        )

        resp = await client.search(

            index="books",

        )

        print(resp)

        await client.close()

    import asyncio

    asyncio.run(test())

from langchain_openai import OpenAIEmbeddings

from app.conf.app_config import EmbeddingConfig, app_config

class EmbeddingClientManager:

    def __init__(self, config: EmbeddingConfig):

        self.client: OpenAIEmbeddings | None = None

        self.config = config

    def _get_url(self):

        return f"http://{self.config.host}:{self.config.port}/v1"

    def init(self):

        self.client = OpenAIEmbeddings(

            model="tei",

            base_url=self._get_url(),

            api_key="dummy",

            check_embedding_ctx_length=False,

        )

embedding_client_manager = EmbeddingClientManager(app_config.embedding)

if __name__ == '__main__':

    embedding_client_manager.init()

    client = embedding_client_manager.client

    text = "Hello, world!"

    embedding = client.embed_query(text)

    print("维度:", len(embedding))

    print("前10维:", embedding[:10])

from sqlalchemy.orm import Session

with Session(engine) as session:

    spongebob = User(

        name="spongebob",

        fullname="Spongebob Squarepants",

        addresses=[Address(email_address="spongebob@sqlalchemy.org")],

    )

    sandy = User(

        name="sandy",

        fullname="Sandy Cheeks",

        addresses=[

            Address(email_address="sandy@sqlalchemy.org"),

            Address(email_address="sandy@squirrelpower.org"),

        ],

    )

    patrick = User(name="patrick", fullname="Patrick Star")

    session.add_all([spongebob, sandy, patrick])

    session.commit()

import asyncio

from sqlalchemy import text, URL

from app.conf.app_config import DBConfig, app_config

from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine, AsyncSession

class MySQLClientManager:

    def __init__(self, config: DBConfig):

        self.engine: AsyncEngine | None = None

        self.config = config

    def _get_url(self):

        return URL.create(

            drivername="mysql+asyncmy",

            username=self.config.user,

            password=self.config.password,

            host=self.config.host,

            port=self.config.port,

            database=self.config.database,

            query={"charset": "utf8mb4"}

        )

    def init(self):

        self.engine = create_async_engine(

            self._get_url(),

            pool_size=50,

        )

    async def close(self):

        await self.engine.dispose()

meta_mysql_client_manager = MySQLClientManager(app_config.db_meta)

dw_mysql_client_manager = MySQLClientManager(app_config.db_dw)

if __name__ == '__main__':

    dw_mysql_client_manager.init()

    dw_engine = dw_mysql_client_manager.engine

    async def test():

        async with AsyncSession(dw_engine) as session:

            sql = "select * from fact_order"

            result = await session.execute(text(sql))

            rows = result.fetchall()

            print(rows)

            print(f"Total rows: {len(rows)}")

            print(rows[0][0])

    asyncio.run(test())

result.mappings()

if __name__ == '__main__':

    dw_mysql_client_manager.init()

    dw_engine = dw_mysql_client_manager.engine

    async def test():

        async with AsyncSession(dw_engine) as session:

            sql = "select * from fact_order"

            result = await session.execute(text(sql))

            result.mappings()

            rows = result.fetchall()

            print(rows[0])

            print(f"Total rows: {len(rows)}")

            print(rows[0].region_id)

    asyncio.run(test())

self.engine = create_async_engine(

    self._get_url(),

    pool_size=50,

)

import asyncio

from sqlalchemy import text, URL

from app.conf.app_config import DBConfig, app_config

from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine, AsyncSession, async_sessionmaker

class MySQLClientManager:

    def __init__(self, config: DBConfig):

        self.engine: AsyncEngine | None = None

        self.session_factory = None

        self.config = config

    def _get_url(self):

        return URL.create(

            drivername="mysql+asyncmy",

            username=self.config.user,

            password=self.config.password,

            host=self.config.host,

            port=self.config.port,

            database=self.config.database,

            query={"charset": "utf8mb4"}

        )

    def init(self):

        self.engine = create_async_engine(

            self._get_url(),

            pool_size=50,

        )

        self.session_factory = async_sessionmaker(

            self.engine,

            expire_on_commit=False,

            autoflush=True

        )

    async def close(self):

        await self.engine.dispose()

meta_mysql_client_manager = MySQLClientManager(app_config.db_meta)

dw_mysql_client_manager = MySQLClientManager(app_config.db_dw)

if __name__ == '__main__':

    dw_mysql_client_manager.init()

    dw_engine = dw_mysql_client_manager.engine

    async def test():

        async with dw_mysql_client_manager.session_factory() as session:

            sql = "select * from fact_order"

            result = await session.execute(text(sql))

            result.mappings()

            rows = result.fetchall()

            print(rows[0])

            print(f"Total rows: {len(rows)}")

            print(rows[0].region_id)

    asyncio.run(test())

from loguru import logger

logger.info("Starting the application...")

logger.warning("Warning: This is a warning message.")

logger.error("Error: This is an error message.")

parser = argparse.ArgumentParser(

    prog="build_meta_knowledge",

    description='Build meta knowledge.',

)

parser.add_argument('-c','--config', default="path/to/config.json", help='Input file')

args = parser.parse_args()

print(f"Using config file: {args.config}")

# _*_ coding : utf-8 _*_

# @Time : 2026/9/21 10:47

# @Author : KarryLiu

# File : meta_config

# @Project : data_agent

from dataclasses import dataclass

from pathlib import Path

from typing import Optional

@dataclass

class ColumnConfig:

    name: str

    role: str

    description: str

    alias: list[str]

    sync: bool

@dataclass

class TableConfig:

    name: str

    role: str

    description: str

    columns: list[ColumnConfig]

@dataclass

class MetricConfig:

    name: str

    description: str

    relevant_columns: list[str]

    alias: list[str]

@dataclass

class MetaConfig:

    tables: Optional[list[TableConfig]] = None

    metrics: Optional[list[MetricConfig]] = None

# _*_ coding : utf-8 _*_

# @Time : 2026/9/19 15:38

# @Author : KarryLiu

# File : build_meta_knowledge

# @Project : data_agent

import argparse

import sys

from pathlib import Path

from app.core.log import logger

from app.services.meta_knowledge_service import MetaKnowledgeService

async def build(config_path: Path):

    meta_knowledge_service = MetaKnowledgeService()

    await meta_knowledge_service.build(config_path)

if __name__ == '__main__':

    parser = argparse.ArgumentParser(

        prog="build_meta_knowledge",

        description='Build meta knowledge.',

    )

    parser.add_argument('-c','--config', default="../../conf/meta_conf.yaml", help='Input file')

    args = parser.parse_args()

    print(f"Using config file: {args.config}")

    import asyncio

    asyncio.run(build(args.config))

import argparse

import sys

from pathlib import Path

from app.clients.mysql_client_mamager import meta_mysql_client_manager, dw_mysql_client_manager

from app.core.log import logger

from app.repositories.mysql.dw.dw_mysql_repo import DWMySQLRepo

from app.repositories.mysql.meta.meta_mysql_repo import MetaMySQLRepo

from app.services.meta_knowledge_service import MetaKnowledgeService

async def build(config_path: Path):

    meta_mysql_client_manager.init()

    dw_mysql_client_manager.init()

    async with (

        meta_mysql_client_manager.session_factory() as meta_session,

        dw_mysql_client_manager.session_factory() as dw_session

    ):

        meta_mysql_repo = MetaMySQLRepo(meta_session)

        dw_mysql_repo = DWMySQLRepo(dw_session)

        meta_knowledge_service = MetaKnowledgeService(

            meta_mysql_repo,

            dw_mysql_repo

        )

        await meta_knowledge_service.build(config_path)

    await meta_mysql_client_manager.close()

    await dw_mysql_client_manager.close()

if __name__ == '__main__':

    parser = argparse.ArgumentParser(

        prog="build_meta_knowledge",

        description='Build meta knowledge.',

    )

    parser.add_argument('-c', '--config', default="../../conf/meta_conf.yaml", help='Input file')

    args = parser.parse_args()

    print(f"Using config file: {args.config}")

    import asyncio

    asyncio.run(build(args.config))

async with (

    meta_mysql_client_manager.session_factory() as meta_session,

    dw_mysql_client_manager.session_factory() as dw_session

):

from sqlalchemy.ext.asyncio import AsyncSession

from app.entities.column_info import ColumnInfo

from app.entities.table_info import TableInfo

from app.repositories.mysql.meta.mappers.column_info_mapper import ColumnInfoMapper

from app.repositories.mysql.meta.mappers.table_info_mapper import tableInfoMapper

class MetaMySQLRepo:

    def __init__(self, session: AsyncSession):

        self.session = session  # MySQL session

    def read(self):

        pass

    def write(self):

        pass

    def save_table_infos(self, table_infos: list[TableInfo]):

        self.session.add_all([tableInfoMapper.to_model(table_info) for table_info in table_infos])

    def save_column_infos(self, column_infos: list[ColumnInfo]):

        self.session.add_all([ColumnInfoMapper.to_model(column_info) for column_info in column_infos])

from sqlalchemy import text

class DWMySQLRepo:

    def __init__(self, session):

        self.session = session  # MySQL session

    def read(self):

        pass

    def write(self):

        pass

    async def get_column_types(self, table_name):

        sql = f"show columns from {table_name};"

        result = await self.session.execute(text(sql))

        result_dict = result.mappings().fetchall()

        return {row['Field']: row['Type'] for row in result_dict}

    async def get_column_examples(self, table_name, col_name):

        sql = f"select distinct {col_name} from {table_name} limit 10;"

        result = await self.session.execute(text(sql))

        result_dict = result.fetchall()

        return [row[0] for row in result_dict]

        # result_dict = result.mappings().fetchall()

        # return [row[col_name] for row in result_dict]

[tableInfoMapper.to_model(table_info) for table_info in table_infos]

{row['Field']: row['Type'] for row in result_dict}

from pathlib import Path

from omegaconf import OmegaConf

from app.conf.meta_config import MetaConfig

from app.entities.column_info import ColumnInfo

from app.entities.table_info import TableInfo

from app.repositories.mysql.dw.dw_mysql_repo import DWMySQLRepo

from app.repositories.mysql.meta.meta_mysql_repo import MetaMySQLRepo

class MetaKnowledgeService:

    def __init__(self, meta_mysql_repo: MetaMySQLRepo, dw_mysql_repo: DWMySQLRepo):

        self.meta_mysql_repo: MetaMySQLRepo = meta_mysql_repo  # MetaMySQLRepo

        self.dw_mysql_repo: DWMySQLRepo = dw_mysql_repo  # DWMySQLRepo

    async def build(self, config_path: Path):

        config_file = config_path

        context = OmegaConf.load(config_file)

        schema = OmegaConf.structured(MetaConfig)

        meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

        # 存在表信息

        if meta_config.tables:

            table_infos: list[TableInfo] = []

            column_infos: list[ColumnInfo] = []

            # 1. 将表信息和字段信息存储到数据库中（table_info，column_info）

            for table in meta_config.tables:

                # table -> table_info

                table_info = TableInfo(

                    id=table.name,

                    name=table.name,

                    description=table.description,

                    role=table.role,

                )

                table_infos.append(table_info)

                # 查询字段类型

                column_types = await self.dw_mysql_repo.get_column_types(table.name)

                for column in table.columns:

                    # 查询字段取值示例

                    column_values = await self.dw_mysql_repo.get_column_examples(table.name, column.name)

                    # table.columns -> column_info

                    column_info = ColumnInfo(

                        id=f"{table.name}.{column.name}",

                        name=column.name,

                        description=column.description,

                        alias=column.alias,

                        type=column_types.get(column.name),

                        role=column.role,

                        examples=column_values,

                        table_id=table.name

                    )

                    column_infos.append(column_info)

            # print(f"table_infos:")

            # for table_info in table_infos:

            #     print(table_info)

            #

            # print("=" * 50)

            # print(f"column_infos:")

            # for column_info in column_infos:

            #     print(column_info)

            # sync with self.meta_mysql_repo.session.begin (): 自动管理事务

            async with self.meta_mysql_repo.session.begin():

                self.meta_mysql_repo.save_table_infos(table_infos)

                self.meta_mysql_repo.save_column_infos(column_infos)

            # 2. 对字段信息（column_info），进行向量化，并存储到向量数据库中（qdrant）

            # 3. 对指定的维度字段，建立全文索引

            pass

        # 存在指标信息

        if meta_config.metrics:

            # 1. 将指标信息存储到数据库中（metric_info, column_metric）

            # 2. 对指标信息进行向量化，并存储到向量数据库中（qdrant）

            pass

if meta_config.tables:

    table_infos: list[TableInfo] = []

    column_infos: list[ColumnInfo] = []