From 2869139ab34e5f9d494bbba69ee234f96ab7d3c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=86=E6=96=8C?= <2669849538@qq.com> Date: Mon, 9 Feb 2026 14:49:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(sync):=20=E6=B7=BB=E5=8A=A0=E8=BE=BE?= =?UTF-8?q?=E6=A2=A6=E6=95=B0=E6=8D=AE=E5=BA=93=E5=90=8C=E6=AD=A5=E6=94=AF?= =?UTF-8?q?=E6=8C=81=20-=20=E5=9C=A8sync=E6=A8=A1=E5=9D=97=E4=B8=AD?= =?UTF-8?q?=E6=96=B0=E5=A2=9EDaMengSync=E7=B1=BB=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E8=BE=BE=E6=A2=A6=E6=95=B0=E6=8D=AE=E5=BA=93=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E5=8A=9F=E8=83=BD=20-=20=E4=BF=AE=E6=94=B9get=5Fsync=5Fclass?= =?UTF-8?q?=E5=87=BD=E6=95=B0=E6=94=AF=E6=8C=81=E6=A0=B9=E6=8D=AE=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E7=B1=BB=E5=9E=8B=E8=BF=94=E5=9B=9E=E7=9B=B8?= =?UTF-8?q?=E5=BA=94=E5=90=8C=E6=AD=A5=E7=B1=BB=20-=20=E6=9B=B4=E6=96=B0Da?= =?UTF-8?q?tabaseDataSourceConfig=E9=85=8D=E7=BD=AE=E7=B1=BB=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E9=80=9A=E7=94=A8=E6=95=B0=E6=8D=AE=E5=BA=93=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E5=8F=82=E6=95=B0=20-=20=E5=B0=86MySQL=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E5=8F=82=E6=95=B0=E4=BB=8E=E4=B8=93=E7=94=A8=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=E9=80=9A=E7=94=A8=E6=95=B0=E6=8D=AE=E5=BA=93=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E5=8F=82=E6=95=B0=20-=20=E5=9C=A8README.md=E4=B8=AD?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E4=B8=8D=E5=90=8C=E6=9E=B6=E6=9E=84=E9=95=9C?= =?UTF-8?q?=E5=83=8F=E4=B8=8B=E8=BD=BD=E8=AF=B4=E6=98=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 13 +- config.py | 49 +++++--- sync/__init__.py | 2 + sync/base_sync.py | 37 +++--- sync/dameng_sync.py | 288 ++++++++++++++++++++++++++++++++++++++++++++ sync/mysql_sync.py | 8 +- 6 files changed, 356 insertions(+), 41 deletions(-) create mode 100644 sync/dameng_sync.py diff --git a/README.md b/README.md index 6932c6e..defa4f3 100644 --- a/README.md +++ b/README.md @@ -178,20 +178,29 @@ ollama pull qwen3-embedding:8b # Embedding模型，用于向量化 注意： + - 确保宿主机上已安装 Docker 和 Docker Compose +- 根据机器架构，下载对应架构的镜像文件后并解压，目前支持 X86 64位和ARM 64架构。 + +``` +X86 64位架构下载 V1.0.0.0.zip，包含镜像 chromadb.tar.gz、soffice.tar.gz、rag-api.tar.gz +ARM 64架构下载 V1.0.0.0-arm64.zip，包含镜像 chromadb-arm64.tar.gz、soffice-arm64.tar.gz、rag-api.tar.gz +``` + +解压后确保： - chromadb.tar.gz、soffice.tar.gz、rag-api.tar.gz 这三个文件必须存在于一个目录下 - .env 文件必须存在于该目录下（按照下面的步骤通过.env.example 生成并修改） - static/ 文件夹必须存在于该目录下 - docker-run.sh 脚本必须存在于该目录下 - nltk/ 文件夹必须存在于该目录下，用于存储 NLTK 数据 - + ```bash # 1. 配置 .env 文件（复制 .env.example 并修改） cp .env.example .env # 编辑 .env 文件，配置 MySQL、Ollama 等连接信息 -# 2. 构建并启动服务 +# 2. 运行docker-run.sh, 构建并启动服务 sudo bash docker-run.sh # 3. 查看服务状态 diff --git a/config.py b/config.py index 238b17f..7d97553 100644 --- a/config.py +++ b/config.py @@ -24,20 +24,24 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig): def __init__( self, name: str, + # 通用数据库连接信息 database: str, table_name: str = "documents", + host: Optional[str] = None, + port: Optional[int] = None, + user: Optional[str] = None, + password: Optional[str] = None, + + db_type: Optional[str] = "mysql", # 新增：数据库类型，支持 "mysql", "dameng" 等 + id_column: str = "id", content_column: str = "content", # 可以是单个列名，或逗号分隔的多个列名 file_column: Optional[str] = None, # 单列名，指向表中的文件标识符字段 title_column: Optional[str] = "title", metadata_columns: Optional[str] = None, - content_separator: str = "\n", # 多个 content 列之间的分隔符 + content_separator: str = "\n", # 多个 content 列拼接时的分隔符 updated_at_column: Optional[str] = None, # 用于增量同步的更新时间字段（可选） - # MySQL connection info (optional, will use .env defaults if not provided) - mysql_host: Optional[str] = None, - mysql_port: Optional[int] = None, - mysql_user: Optional[str] = None, - mysql_password: Optional[str] = None, + # 文件源配置 file_source_type: Optional[str] = None, # 可选值: "api", "filesystem", "scp" file_system_base_path: Optional[str] = None, # 文件系统基础路径 @@ -49,8 +53,16 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig): scp_key_path: Optional[str] = None ): super().__init__(name, "database") - self.database = database # MySQL数据库名称 + + # 通用数据库连接信息 + self.db_type = db_type # 数据库类型，mysql or dameng + self.database = database # 数据库名称 self.table_name = table_name + self.host = host + self.port = port + self.user = user + self.password = password + self.id_column = id_column # 支持多个 content_column（逗号分隔） if content_column: @@ -64,13 +76,7 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig): self.metadata_columns = metadata_columns self.content_separator = content_separator # 多个列之间的分隔符 self.updated_at_column = updated_at_column # 更新时间字段（用于增量同步） - - # MySQL connection info (optional, will fallback to .env defaults) - self.mysql_host = mysql_host - self.mysql_port = mysql_port - self.mysql_user = mysql_user - self.mysql_password = mysql_password - + # 文件源配置 self.file_source_type = file_source_type self.file_system_base_path = file_system_base_path @@ -163,6 +169,7 @@ class Settings(BaseSettings): NLTK_DATA: str = _DEFAULT_NLTK + # 文件下载接口地址，根据实际环境进行修改 FILE_DOWNLOAD_BASE_URL: str = "http://172.20.32.184:8000/api/file/open/downloadByIdentifier" # 添加 SOFFICE 配置 @@ -216,8 +223,15 @@ class Settings(BaseSettings): # Create database data source configs.append(DatabaseDataSourceConfig( name=name, # 使用数据库表中的name列 + # 通用数据库连接信息 + db_type= ds_config.get('db_type', 'mysql'), # Default to 'mysql' database=ds_config['database'], # Required field - table_name=ds_config.get('table_name', 'documents'), # Default to 'documents' + table_name=ds_config.get('table_name', 'documents'), + host=ds_config.get('host', None), + port=ds_config.get('port', None), + user=ds_config.get('user', None), + password=ds_config.get('password', None), + id_column=ds_config.get('id_column', 'id'), # Default to 'id' content_column=ds_config.get('content_column', 'content'), # Default to 'content' file_column=ds_config.get('file_column', None), @@ -225,11 +239,6 @@ class Settings(BaseSettings): metadata_columns=ds_config.get('metadata_columns', None), content_separator=ds_config.get('content_separator', '\n'), updated_at_column=ds_config.get('updated_at_column', None), - # MySQL connection info - mysql_host=ds_config.get('mysql_host', None), - mysql_port=ds_config.get('mysql_port', None), - mysql_user=ds_config.get('mysql_user', None), - mysql_password=ds_config.get('mysql_password', None), # 文件源配置 file_source_type=ds_config.get('file_source_type', None), file_system_base_path=ds_config.get('file_system_base_path', None), diff --git a/sync/__init__.py b/sync/__init__.py index 305bf76..08f5c44 100644 --- a/sync/__init__.py +++ b/sync/__init__.py @@ -1,5 +1,6 @@ """Synchronization modules for all data sources""" from .base_sync import BaseSync, get_sync_class +from .dameng_sync import DaMengSync from .mysql_sync import MySQLSync from .folder_sync import FolderSync @@ -7,5 +8,6 @@ __all__ = [ 'BaseSync', 'get_sync_class', 'MySQLSync', + 'DaMengSync', 'FolderSync' ] diff --git a/sync/base_sync.py b/sync/base_sync.py index 21541de..94b6de5 100644 --- a/sync/base_sync.py +++ b/sync/base_sync.py @@ -269,36 +269,43 @@ class BaseSync(ABC): missing = [] for config in configs: - # Get the appropriate sync class based on data source type - sync_class = get_sync_class(config.type) + # 如果是数据库类型，使用db_type参数 + if config.type == 'database': + sync_class = get_sync_class(config.type, getattr(config, 'db_type', 'mysql')) + else: + sync_class = get_sync_class(config.type) if not sync_class.check_data_source_exists(config): missing.append(config.name) return len(missing) == 0, missing -def get_sync_class(source_type: str) -> type[BaseSync]: +def get_sync_class(source_type: str, db_type: str = 'mysql') -> type[BaseSync]: """ Get the appropriate sync class based on data source type - + Args: source_type: Type of data source (database, folder) - + db_type: Type of database (mysql, dameng, etc.) - only used when source_type is 'database' + Returns: Sync class corresponding to the data source type - + Raises: ValueError: If source type is not supported """ from sync.mysql_sync import MySQLSync from sync.folder_sync import FolderSync - - sync_classes = { - 'database': MySQLSync, # Currently only MySQL, but can be extended - 'folder': FolderSync - } - - if source_type not in sync_classes: + from sync.dameng_sync import DaMengSync + + if source_type == 'database': + # 根据数据库类型选择相应的同步类 + db_type_lower = db_type.lower() + if db_type_lower == 'dameng': + return DaMengSync + else: # 默认为mysql + return MySQLSync + elif source_type == 'folder': + return FolderSync + else: raise ValueError(f"Unsupported data source type: {source_type}") - - return sync_classes[source_type] diff --git a/sync/dameng_sync.py b/sync/dameng_sync.py new file mode 100644 index 0000000..13042a0 --- /dev/null +++ b/sync/dameng_sync.py @@ -0,0 +1,288 @@ +"""Dameng database synchronization implementation""" +import urllib +import dmPython # 达梦数据库驱动 +import requests +from typing import List, Dict, Optional, Tuple, Any, Set +from datetime import datetime +from loguru import logger +from config import DatabaseDataSourceConfig as DatabaseConfig, settings +from rag.file_parser import FileParser +from sync.base_sync import BaseSync + + +class DaMengSync(BaseSync): + """Handle synchronization between Dameng database and ChromaDB""" + + def __init__(self, db_config: DatabaseConfig, vector_store_manager=None): + """ + Initialize DaMeng sync with database configuration + + Args: + db_config: DatabaseConfig object containing database table info + """ + super().__init__(db_config) + self.db_config = self.config + self.vector_store_manager = vector_store_manager + self.file_parser = FileParser() + self.connection = None + + def _connect(self): + """Create Dameng database connection""" + # Get connection parameters from config or use defaults + host = self.db_config.host or settings.HOST + port = self.db_config.port or settings.PORT + user = self.db_config.user or settings.USER + password = self.db_config.password or settings.PASSWORD + # database = self.db_config.database or settings.DATABASE + + # Log connection info (without password) + logger.info(f"Connecting to DaMeng Schema: {user}") + logger.debug(f"DaMeng connection details: host={host}, port={port}, user={user}") + + # Create connection + self.connection = dmPython.connect( + user = user, + password = password, + server = host, + port = port + ) + + def fetch_all_documents(self, last_sync_time=None) -> List[Dict[str, Any]]: + """ + Fetch documents from the DaMeng database + + Args: + last_sync_time: Last synchronization time (for incremental sync) + + Returns: + List of documents + """ + self._connect() + cursor = self.connection.cursor() + try: + # Build query with all columns + columns = [ + self.db_config.id_column, + self.db_config.title_column, + self.db_config.content_column, + self.db_config.file_column, + self.db_config.updated_at_column + ] if self.db_config.updated_at_column else [ + self.db_config.id_column, + self.db_config.title_column, + self.db_config.content_column, + self.db_config.file_column + ] + + # Remove duplicates and None values + columns = list(set([col for col in columns if col])) + + # Build the query + query = f"SELECT {', '.join(columns)} FROM {self.db_config.table_name}" + + # Add incremental sync condition if applicable + params = [] + if last_sync_time and self.db_config.updated_at_column: + query += f" WHERE {self.db_config.updated_at_column} > ?" + params.append(last_sync_time) + + logger.debug(f"DaMeng query: {query}, params: {params}") + cursor.execute(query, params) + + # Get column names from cursor description + column_names = [desc[0] for desc in cursor.description] + + # Parse results + documents = [] + for row in cursor.fetchall(): + # Convert row tuple to dict + row_dict = dict(zip(column_names, row)) + + # Handle file content if file_column is specified + if self.db_config.file_column and row_dict.get(self.db_config.file_column): + # Extract file path from the file column + file_path = row_dict[self.db_config.file_column] + + # Load file content if file source is configured + if self.db_config.file_source_type: + file_content = self._load_file_content(file_path) + if file_content: + row_dict[self.db_config.content_column] = file_content + + # Generate unique document ID + record_id = str(row_dict[self.db_config.id_column]) + doc_id = self.generate_doc_id(record_id) + row_dict['id'] = doc_id + + # Check if document has already been synced + if self.vector_store_manager and self.vector_store_manager.document_exists(doc_id + "_chunk_0"): + # If document is already synced, check if it's been modified since last sync + if last_sync_time and self.db_config.updated_at_column and row_dict.get( + self.db_config.updated_at_column): + # Skip if not modified since last sync + if row_dict[self.db_config.updated_at_column] <= last_sync_time: + continue + elif last_sync_time: + # No updated_at column, skip since we can't determine if it's been modified + continue + + documents.append(row_dict) + + return documents + finally: + cursor.close() + + def fetch_new_documents(self, last_sync_time=None) -> List[Dict[str, Any]]: + """ + Fetch new/updated documents from the Dameng database since last sync time + + Args: + last_sync_time: Last synchronization time + + Returns: + List of new/updated documents + """ + # Call fetch_all_documents which now handles document existence checks + return self.fetch_all_documents(last_sync_time) + + def get_synced_document_ids(self) -> Set[str]: + """ + Get IDs of all documents in the Dameng database + + Returns: + Set of document IDs + """ + cursor = self.connection.cursor() + try: + query = f"SELECT {self.db_config.id_column} FROM {self.db_config.table_name}" + cursor.execute(query) + column_names = [desc[0] for desc in cursor.description] + id_index = column_names.index(self.db_config.id_column) + return {str(row[id_index]) for row in cursor.fetchall()} + finally: + cursor.close() + + def generate_doc_id(self, record_id: str) -> str: + """ + Generate a unique document ID for Dameng records + + Args: + record_id: ID of the record in the database + + Returns: + Unique document ID based on user, table, and record ID + """ + # 为达梦数据库记录生成唯一的文档 ID + return f"{self.db_config.user}_dameng_{self.db_config.table_name}_{record_id}" + + def doc_to_llamaindex_doc(self, doc: Dict) -> 'Document': + """ + Convert Dameng document to LlamaIndex Document + + Args: + doc: Dameng document dictionary + + Returns: + LlamaIndex Document object + """ + from llama_index.core import Document + + # 处理多个 content 列（支持合并多个列的内容） + if self.db_config: + # 使用配置的多个 content 列 + content_columns = self.db_config.content_columns + content_separator = self.db_config.content_separator + else: + # 向后兼容：使用单个 content_column + content_columns = ["content"] # Default to "content" column + content_separator = "\n" + + # 合并所有 content 列的内容 + content_parts = [] + for col in content_columns: + col_value = doc.get(col, "") + if col_value: + content_parts.append(str(col_value)) + + # 用指定的分隔符连接多个列的内容 + if content_parts: + if content_separator: + content = content_separator.join(content_parts) + else: + content = " ".join(content_parts) # 如果没有指定分隔符，使用空格 + else: + content = "" + + title = doc.get('title', "") + doc_id = doc.get('id', "") + + # Build metadata + metadata = { + "doc_id": doc_id, + "source": "dameng", + # Dameng中没有database的概念 + "schema": self.db_config.user, + "table": self.db_config.table_name + } + + if title: + metadata["title"] = title + + # Create Document + return Document( + text=content, + id_=doc_id, + metadata=metadata + ) + + + @staticmethod + def check_data_source_exists(config: DatabaseConfig) -> bool: + """ + Check if the DaMeng table exists and is accessible + + Args: + config: Database configuration + + Returns: + True if table exists and is accessible, False otherwise + """ + temp_connection = None + try: + # Get connection parameters + host = config.host or settings.HOST + port = config.port or settings.PORT + user = config.user or settings.USER + password = config.password or settings.PASSWORD + + # Create connection + temp_connection = dmPython.connect( + user=user, + password=password, + server=host, + port=port + ) + + # Check if table exists + cursor = temp_connection.cursor() + cursor.execute( + # "SELECT TABLE_NAME FROM USER_TABLES WHERE TABLE_NAME = ?", + "SELECT COUNT(*) FROM ALL_TABLES WHERE OWNER = ? AND TABLE_NAME = ?", + [config.user.upper(), config.table_name.upper()] # DM 模式名、表名默认大写 + ) + result = cursor.fetchone() + cursor.close() + + # 判断查询结果 + if result and result[0] > 0: + return True # 表存在 + else: + logger.error(f"Table {config.table_name} does not exist in database {config.database}") + return False # 表不存在 + + except Exception as e: + logger.error(f"Error checking Dameng data source: {e}") + return False + finally: + if temp_connection: + temp_connection.close() diff --git a/sync/mysql_sync.py b/sync/mysql_sync.py index 7a0c24a..cc79bb5 100644 --- a/sync/mysql_sync.py +++ b/sync/mysql_sync.py @@ -29,10 +29,10 @@ class MySQLSync(BaseSync): def _connect(self): """Create MySQL connection""" # Get connection parameters from config or use defaults - host = self.db_config.mysql_host or settings.MYSQL_HOST - port = self.db_config.mysql_port or settings.MYSQL_PORT - user = self.db_config.mysql_user or settings.MYSQL_USER - password = self.db_config.mysql_password or settings.MYSQL_PASSWORD + host = self.db_config.host or settings.HOST + port = self.db_config.port or settings.PORT + user = self.db_config.user or settings.USER + password = self.db_config.password or settings.PASSWORD # Log connection info (without password) logger.info(f"Connecting to MySQL database: {self.db_config.database}")