diff --git a/.gitignore b/.gitignore index 24cc791..d94f0cb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,4 @@ -!install/ +install/ # Python __pycache__/ *.py[cod] diff --git a/Dockerfile b/Dockerfile index 9a5e027..5ca09e6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -26,6 +26,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ gcc \ g++ \ curl \ + libssl-dev \ + libcrypto++-dev \ + libgmp-dev \ && rm -rf /var/lib/apt/lists/* # 配置 pip 镜像源（加速 Python 包安装） diff --git a/README.md b/README.md index 2fdfa40..aba4f29 100644 --- a/README.md +++ b/README.md @@ -180,19 +180,27 @@ 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/api/main.py b/api/main.py index abf46ed..0f64775 100644 --- a/api/main.py +++ b/api/main.py @@ -1343,7 +1343,10 @@ async def create_config(config: Dict[str, Any]): # 数据库配置需要：主机、端口、数据库名、表名 if not config.get("mysql_host") or not config.get("mysql_port") or not config.get("database") or not config.get("table_name"): raise HTTPException(status_code=400, detail="数据库配置必须包含主机、端口、数据库名和表名") + # 添加数据库类型到唯一标识符 + db_type = config.get("db_type", "mysql").lower() unique_id_parts.extend([ + db_type, config["mysql_host"].lower(), str(config["mysql_port"]), config["database"].lower(), @@ -1383,8 +1386,9 @@ async def create_config(config: Dict[str, Any]): # Check if it's the same type if existing_config_data.get('type') == config_type: if config_type == 'database': - # For database configs, same source means same host, port, and db name - if (existing_config_data.get('mysql_host') == config.get('mysql_host') and + # For database configs, same source means same db type, host, port, and db name + if (existing_config_data.get('db_type', 'mysql') == config.get('db_type', 'mysql') and + existing_config_data.get('mysql_host') == config.get('mysql_host') and existing_config_data.get('mysql_port') == config.get('mysql_port') and existing_config_data.get('database') == config.get('database')): raise HTTPException( @@ -1443,10 +1447,11 @@ async def create_config(config: Dict[str, Any]): metadata_columns=config.get("metadata_columns"), content_separator=config.get("content_separator"), updated_at_column=config.get("updated_at_column"), - mysql_host=config.get("mysql_host"), - mysql_port=config.get("mysql_port"), - mysql_user=config.get("mysql_user"), - mysql_password=config.get("mysql_password"), + host=config.get("mysql_host"), + port=config.get("mysql_port"), + user=config.get("mysql_user"), + password=config.get("mysql_password"), + db_type=config.get("db_type", "mysql"), file_source_type=config.get("file_source_type"), file_system_base_path=config.get("file_system_base_path"), scp_host=config.get("scp_host"), @@ -1587,12 +1592,13 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]): # 根据不同类型的配置生成有意义的ID if config_type == "database": - # 数据库配置：使用数据库名和表名生成ID + # 数据库配置：使用数据库类型、数据库名和表名生成ID if not config.get("database"): raise HTTPException(status_code=400, detail="Database name is required for database configuration") if not config.get("table_name"): raise HTTPException(status_code=400, detail="Table name is required for database configuration") - new_config_id = f"{config_type}_{config['database'].lower()}_{config['table_name'].lower()}" + db_type = config.get("db_type", "mysql").lower() + new_config_id = f"{config_type}_{db_type}_{config['database'].lower()}_{config['table_name'].lower()}" elif config_type == "folder": # 文件夹配置：根据是否有host字段区分本地和远程 if not config.get("folder_path"): @@ -1648,10 +1654,11 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]): metadata_columns=config.get("metadata_columns"), content_separator=config.get("content_separator"), updated_at_column=config.get("updated_at_column"), - mysql_host=config.get("mysql_host"), - mysql_port=config.get("mysql_port"), - mysql_user=config.get("mysql_user"), - mysql_password=config.get("mysql_password"), + host=config.get("mysql_host"), + port=config.get("mysql_port"), + user=config.get("mysql_user"), + password=config.get("mysql_password"), + db_type=config.get("db_type", "mysql"), file_source_type=config.get("file_source_type"), file_system_base_path=config.get("file_system_base_path"), scp_host=config.get("scp_host"), @@ -1751,6 +1758,7 @@ class DatabaseConnectionParams(BaseModel): port: int = Field(default=3306, description="Database port") username: str = Field(..., description="Database username") password: str = Field(..., description="Database password") + db_type: str = Field(default="mysql", description="Database type: mysql or dameng") class DatabaseParams(BaseModel): """Database parameters""" @@ -1759,6 +1767,7 @@ class DatabaseParams(BaseModel): username: str = Field(..., description="Database username") password: str = Field(..., description="Database password") database: str = Field(..., description="Database name") + db_type: str = Field(default="mysql", description="Database type: mysql or dameng") class TableParams(BaseModel): """Table parameters""" @@ -1768,6 +1777,7 @@ class TableParams(BaseModel): password: str = Field(..., description="Database password") database: str = Field(..., description="Database name") table_name: str = Field(..., description="Table name") + db_type: str = Field(default="mysql", description="Database type: mysql or dameng") @@ -1775,7 +1785,7 @@ class TableParams(BaseModel): @app.post("/database/databases") async def get_databases(params: DatabaseConnectionParams): """ - Get list of databases from MySQL server + Get list of databases from database server Args: params: Database connection parameters @@ -1784,34 +1794,53 @@ async def get_databases(params: DatabaseConnectionParams): List of available databases """ try: - # Direct database connection - connection = mysql.connector.connect( - host=params.host, - port=params.port, - user=params.username, - password=params.password - ) - - cursor = connection.cursor() - cursor.execute("SHOW DATABASES") - - databases = [] - for (database_name,) in cursor: - # Skip system databases - if database_name not in ['information_schema', 'mysql', 'performance_schema', 'sys']: - databases.append(database_name) - - cursor.close() - connection.close() + if params.db_type == "dameng": + # DaMeng database connection + import dmPython + connection = dmPython.connect( + user=params.username, + password=params.password, + server=params.host, + port=params.port + ) + + cursor = connection.cursor() + # Get schemas for DaMeng (similar to databases in MySQL) + cursor.execute("SELECT NAME AS SCHEMA_NAME FROM SYSOBJECTS WHERE TYPE$ = 'SCH';") + + databases = [] + for (schema_name,) in cursor: + databases.append(schema_name) + + cursor.close() + connection.close() + + else: + # MySQL database connection + connection = mysql.connector.connect( + host=params.host, + port=params.port, + user=params.username, + password=params.password + ) + + cursor = connection.cursor() + cursor.execute("SHOW DATABASES") + + databases = [] + for (database_name,) in cursor: + # Skip system databases + if database_name not in ['information_schema', 'mysql', 'performance_schema', 'sys']: + databases.append(database_name) + + cursor.close() + connection.close() return {"databases": databases} - except mysql.connector.Error as e: + except Exception as e: logger.error(f"Database connection error: {e}") raise HTTPException(status_code=500, detail=f"Database connection failed: {str(e)}") - except Exception as e: - logger.error(f"Error getting databases: {e}") - raise HTTPException(status_code=500, detail=f"Error getting databases: {str(e)}") @app.post("/database/tables") @@ -1826,33 +1855,52 @@ async def get_tables(params: DatabaseParams): List of tables in the specified database """ try: - # Direct database connection - connection = mysql.connector.connect( - host=params.host, - port=params.port, - user=params.username, - password=params.password, - database=params.database - ) - - cursor = connection.cursor() - cursor.execute("SHOW TABLES") - - tables = [] - for (table_name,) in cursor: - tables.append(table_name) - - cursor.close() - connection.close() + if params.db_type == "dameng": + # DaMeng database connection + import dmPython + connection = dmPython.connect( + user=params.username, + password=params.password, + server=params.host, + port=params.port + ) + + cursor = connection.cursor() + # Get tables for DaMeng + cursor.execute(f"SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER = '{params.database.upper()}'") + + tables = [] + for (table_name,) in cursor: + tables.append(table_name) + + cursor.close() + connection.close() + + else: + # MySQL database connection + connection = mysql.connector.connect( + host=params.host, + port=params.port, + user=params.username, + password=params.password, + database=params.database + ) + + cursor = connection.cursor() + cursor.execute("SHOW TABLES") + + tables = [] + for (table_name,) in cursor: + tables.append(table_name) + + cursor.close() + connection.close() return {"tables": tables} - except mysql.connector.Error as e: + except Exception as e: logger.error(f"Database connection error: {e}") raise HTTPException(status_code=500, detail=f"Database connection failed: {str(e)}") - except Exception as e: - logger.error(f"Error getting tables: {e}") - raise HTTPException(status_code=500, detail=f"Error getting tables: {str(e)}") @app.post("/database/table-structure") @@ -1867,40 +1915,66 @@ async def get_table_structure(params: TableParams): Table structure with column details """ try: - # Direct database connection - connection = mysql.connector.connect( - host=params.host, - port=params.port, - user=params.username, - password=params.password, - database=params.database - ) - - cursor = connection.cursor() - cursor.execute(f"DESCRIBE {params.table_name}") - - columns = [] - for (field, type, null, key, default, extra) in cursor: - columns.append({ - "name": field, - "type": type, - "null": null, - "key": key, - "default": default, - "extra": extra - }) - - cursor.close() - connection.close() + if params.db_type == "dameng": + # DaMeng database connection + import dmPython + connection = dmPython.connect( + user=params.username, + password=params.password, + server=params.host, + port=params.port + ) + + cursor = connection.cursor() + # Get table structure for DaMeng + cursor.execute(f"SELECT COLUMN_NAME, DATA_TYPE, NULLABLE, COLUMN_ID FROM ALL_TAB_COLUMNS WHERE OWNER = '{params.database.upper()}' AND TABLE_NAME = '{params.table_name.upper()}' ORDER BY COLUMN_ID") + + columns = [] + for (column_name, data_type, nullable, column_id) in cursor: + columns.append({ + "name": column_name, + "type": data_type, + "null": "YES" if nullable == 'Y' else "NO", + "key": "", # DaMeng doesn't return key info in this query + "default": None, + "extra": "" + }) + + cursor.close() + connection.close() + + else: + # MySQL database connection + connection = mysql.connector.connect( + host=params.host, + port=params.port, + user=params.username, + password=params.password, + database=params.database + ) + + cursor = connection.cursor() + cursor.execute(f"DESCRIBE {params.table_name}") + + columns = [] + for (field, type, null, key, default, extra) in cursor: + columns.append({ + "name": field, + "type": type, + "null": null, + "key": key, + "default": default, + "extra": extra + }) + + cursor.close() + connection.close() return {"columns": columns} - except mysql.connector.Error as e: + except Exception as e: logger.error(f"Database connection error: {e}") raise HTTPException(status_code=500, detail=f"Database connection failed: {str(e)}") - except Exception as e: - logger.error(f"Error getting table structure: {e}") - raise HTTPException(status_code=500, detail=f"Error getting table structure: {str(e)}") 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/requirements.txt b/requirements.txt index 822e74f..76545b4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -17,6 +17,7 @@ pymysql==1.1.0 sqlalchemy>=2.0.40 cryptography>=46.0.3 mysql-connector-python>=9.5.0 +dmpython==2.5.30 # Utilities python-dotenv==1.0.0 diff --git a/static/config/index.html b/static/config/index.html index 8688ea8..50a2af2 100644 --- a/static/config/index.html +++ b/static/config/index.html @@ -4,8 +4,8 @@ RAG 配置管理 - - + +
@@ -77,6 +77,13 @@
+
diff --git a/static/config/script.js b/static/config/script.js index b4c1071..57509a1 100644 --- a/static/config/script.js +++ b/static/config/script.js @@ -290,6 +290,13 @@ function generateConfigForm(config) { const connectionSection = document.createElement('div'); connectionSection.innerHTML = `

数据库连接配置

+
+ + +
@@ -307,8 +314,6 @@ function generateConfigForm(config) {
- -
@@ -478,6 +483,7 @@ function bindDatabaseEvents() { const port = parseInt(document.getElementById('formMysqlPort').value); const username = document.getElementById('formMysqlUser').value; const password = document.getElementById('formMysqlPassword').value; + const dbType = document.getElementById('formDbType').value; if (!host || !username) { alert('请填写主机地址和用户名'); @@ -530,6 +536,7 @@ function bindDatabaseEvents() { port, username, password, + db_type: dbType, ...sshConfig }) }); @@ -558,6 +565,7 @@ function bindDatabaseEvents() { const port = parseInt(document.getElementById('formMysqlPort').value); const username = document.getElementById('formMysqlUser').value; const password = document.getElementById('formMysqlPassword').value; + const dbType = document.getElementById('formDbType').value; if (!host || !username) { alert('请填写主机地址和用户名'); @@ -608,6 +616,7 @@ function bindDatabaseEvents() { port, username, password, + db_type: dbType, ...sshConfig }) }); @@ -651,6 +660,7 @@ function bindDatabaseEvents() { const username = document.getElementById('formMysqlUser').value; const password = document.getElementById('formMysqlPassword').value; const database = document.getElementById('formDatabase').value; + const dbType = document.getElementById('formDbType').value; if (!host || !username || !database) { alert('请填写完整的数据库连接信息并选择数据库'); @@ -702,6 +712,7 @@ function bindDatabaseEvents() { username, password, database, + db_type: dbType, ...sshConfig }) }); @@ -746,6 +757,7 @@ function bindDatabaseEvents() { const password = document.getElementById('formMysqlPassword').value; const database = document.getElementById('formDatabase').value; const table_name = document.getElementById('formTableName').value; + const dbType = document.getElementById('formDbType').value; if (!host || !username || !database || !table_name) { alert('请填写完整的数据库连接信息并选择数据库和表'); @@ -798,6 +810,7 @@ function bindDatabaseEvents() { password, database, table_name, + db_type: dbType, ...sshConfig }) }); @@ -977,7 +990,8 @@ async function fetchTableStructure(config) { username: config.mysql_user, password: config.mysql_password || '', database: config.database, - table_name: config.table_name + table_name: config.table_name, + db_type: config.db_type || 'mysql' }) }); @@ -1010,6 +1024,21 @@ function showConfigScreen() { // 显示添加配置模态框 function showAddConfigModal() { document.getElementById('addConfigModal').classList.add('show'); + + // 添加配置类型选择的事件监听器 + const configTypeSelect = document.getElementById('configType'); + const dbTypeGroup = document.getElementById('dbTypeGroup'); + + configTypeSelect.addEventListener('change', function() { + if (this.value === 'database') { + dbTypeGroup.style.display = 'block'; + } else { + dbTypeGroup.style.display = 'none'; + } + }); + + // 触发一次change事件，确保初始状态正确 + configTypeSelect.dispatchEvent(new Event('change')); } // 隐藏添加配置模态框 @@ -1042,7 +1071,8 @@ async function handleAddConfigDirectly() { database: '', table_name: '', id_column: '', - content_column: '' + content_column: '', + db_type: 'mysql' }; } else if (configType === 'folder') { tempConfig = { @@ -1090,6 +1120,7 @@ async function handleAddConfig(event) { const formData = new FormData(event.target); const configType = formData.get('type'); const configName = formData.get('name'); + const dbType = formData.get('dbType') || 'mysql'; // 根据配置类型创建不同的配置对象 let configData; @@ -1104,7 +1135,8 @@ async function handleAddConfig(event) { database: '', table_name: '', id_column: '', - content_column: '' + content_column: '', + db_type: dbType }; } else if (configType === 'folder') { configData = { @@ -1379,6 +1411,7 @@ function collectFormData() { formData.mysql_port = parseInt(document.getElementById('formMysqlPort').value); formData.mysql_user = document.getElementById('formMysqlUser').value; formData.mysql_password = document.getElementById('formMysqlPassword').value; + formData.db_type = document.getElementById('formDbType').value; 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}") diff --git a/sync_service.py b/sync_service.py index e5c5051..70a1998 100644 --- a/sync_service.py +++ b/sync_service.py @@ -23,7 +23,10 @@ class SyncService: try: # 直接调用具体同步类的 check_data_source_exists 方法，以便捕获详细的错误信息 - sync_class = get_sync_class(self.source_config.type) + if self.source_config.type == 'database': + sync_class = get_sync_class(self.source_config.type, getattr(self.source_config, 'db_type', 'mysql')) + else: + sync_class = get_sync_class(self.source_config.type) sync_class.check_data_source_exists(self.source_config) logger.info(f"✓ Data source {self.source_name} exists") @@ -48,7 +51,10 @@ class SyncService: # Initialize syncer for this data source try: # Get the appropriate sync class based on data source type - sync_class = get_sync_class(self.source_config.type) + if self.source_config.type == 'database': + sync_class = get_sync_class(self.source_config.type, getattr(self.source_config, 'db_type', 'mysql')) + else: + sync_class = get_sync_class(self.source_config.type) # Create a new syncer instance for this data source self.syncer = sync_class(self.source_config, self.vector_store_manager)