Kexing
/
RAG
forked from Nigel/RAG
0
0
Fork 1
Code Issues Pull Requests 6 Actions Packages Projects Releases Wiki Activity

合并上游master分支，解决.gitignore文件冲突

This commit is contained in:
linlin 2026-02-23 16:56:21 +08:00
parent 6c14f69dac f47f9edb7f
commit 31c3f4bc08
13 changed files with 576 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,4 +1,4 @@
!install/
install/
# Python
__pycache__/
*.py[cod]

3
Dockerfile
View File

@ -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 包安装）

12
README.md
View File

@ -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 数据
</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. 查看服务状态

248
api/main.py
View File

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

49
config.py
View File

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

1
requirements.txt
View File

@ -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

11
static/config/index.html
View File

@ -4,8 +4,8 @@
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>RAG 配置管理</title>
<link rel="stylesheet" href="/static/config/style.css?v=202601251455">
<script src="/static/config/script.js?v=202601251455"></script>
<link rel="stylesheet" href="/static/config/style.css?v=202602101450">
<script src="/static/config/script.js?v=202602101450"></script>
</head>
<body>
<div class="config-container">
@ -77,6 +77,13 @@
<option value="folder">文件夹 (folder)</option>
</select>
</div>
<div class="form-group" id="dbTypeGroup" style="display: none;">
<label for="dbType">数据库类型</label>
<select id="dbType" name="dbType">
<option value="mysql">MySQL</option>
<option value="dameng">达梦数据库</option>
</select>
</div>
<div class="modal-footer">
<button type="button" class="btn secondary" id="cancelAddBtn">取消</button>
<button type="submit" class="btn primary">创建</button>

43
static/config/script.js
View File

@ -290,6 +290,13 @@ function generateConfigForm(config) {
const connectionSection = document.createElement('div');
connectionSection.innerHTML = `
<h3 class="section-title">数据库连接配置</h3>
<div class="form-group">
<label for="formDbType">数据库类型 <span class="required">*</span></label>
<select id="formDbType" required>
<option value="mysql" ${config.db_type === 'mysql' ? 'selected' : config.db_type === undefined ? 'selected' : ''}>MySQL</option>
<option value="dameng" ${config.db_type === 'dameng' ? 'selected' : ''}>达梦数据库</option>
</select>
</div>
<div class="form-group">
<label for="formMysqlHost">主机地址 <span class="required">*</span></label>
<input type="text" id="formMysqlHost" value="${config.mysql_host || ''}" required>
@ -307,8 +314,6 @@ function generateConfigForm(config) {
<input type="password" id="formMysqlPassword" value="${config.mysql_password || ''}" required>
</div>
<div class="form-group">
<button type="button" id="testConnectionBtn" class="btn primary" style="margin-right: 10px;">🔌 测试连接</button>
<button type="button" id="getDatabasesBtn" class="btn primary">📋 获取数据库列表</button>
@ -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;

2
sync/__init__.py
View File

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

37
sync/base_sync.py
View File

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

288
sync/dameng_sync.py Normal file
View File

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

8
sync/mysql_sync.py
View File

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

10
sync_service.py
View File

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