feat(sync): 添加达梦数据库同步支持

- 在sync模块中新增DaMengSync类实现达梦数据库同步功能
- 修改get_sync_class函数支持根据数据库类型返回相应同步类
- 更新DatabaseDataSourceConfig配置类支持通用数据库连接参数
- 将MySQL连接参数从专用改为通用数据库连接参数
- 在README.md中添加不同架构镜像下载说明

README.md

@@ -178,20 +178,29 @@ ollama pull qwen3-embedding:8b    # Embedding模型，用于向量化
 
 <strong>
 注意：
+
 - 确保宿主机上已安装 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 数据
-</strong>
+  </strong>
 
 ```bash
 # 1. 配置 .env 文件（复制 .env.example 并修改）
 cp .env.example .env
 # 编辑 .env 文件，配置 MySQL、Ollama 等连接信息
 
-# 2. 构建并启动服务
+# 2. 运行docker-run.sh, 构建并启动服务
 sudo bash docker-run.sh
 
 # 3. 查看服务状态

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),

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'
 ]

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]

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()

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}")