diff --git a/.env.example b/.env.example
index 882ef73..af1c18d 100644
--- a/.env.example
+++ b/.env.example
@@ -45,7 +45,6 @@ MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=forgeplus
-MYSQL_CHARSET=utf8mb4
# 多数据库配置（可选）
# 方式1: 使用配置文件（推荐）
@@ -77,7 +76,6 @@ MYSQL_CHARSET=utf8mb4
# "mysql_port": 3306,
# "mysql_user": "remote_user",
# "mysql_password": "remote_password",
-# "mysql_charset": "utf8mb4",
# "comment": "使用独立的 MySQL 连接配置（不同的服务器）"
# }
# ]
diff --git a/api/main.py b/api/main.py
index 6da91ff..6a7fd38 100644
--- a/api/main.py
+++ b/api/main.py
@@ -14,11 +14,11 @@ from fastapi.responses import StreamingResponse, Response, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
from pydantic import BaseModel, Field
-from typing import Optional, List
+from typing import Optional, List, Dict, Any
from loguru import logger
from config import settings
from rag import VectorStoreManager, RAGEngine, FileParser, DocumentProcessor
-from sync_service import SyncService
+from sync_service import SyncServiceManager
import requests
from datetime import datetime
import time
@@ -40,7 +40,7 @@ import hashlib
# Global instances
vector_store_manager: Optional[VectorStoreManager] = None
rag_engine: Optional[RAGEngine] = None
-sync_service: Optional[SyncService] = None
+sync_manager: Optional[SyncServiceManager] = None
file_parser: Optional[FileParser] = None
document_processor: Optional[DocumentProcessor] = None
auto_sync_task = None # Keep reference to prevent garbage collection
@@ -49,7 +49,7 @@ auto_sync_task = None # Keep reference to prevent garbage collection
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup and shutdown events"""
- global vector_store_manager, rag_engine, sync_service, file_parser, document_processor, auto_sync_task
+ global vector_store_manager, rag_engine, sync_manager, file_parser, document_processor, auto_sync_task
# Startup
try:
@@ -93,66 +93,26 @@ async def lifespan(app: FastAPI):
# This allows API to start immediately even if MySQL connection fails
async def init_sync_service():
"""Initialize sync service and start sync in background"""
- global sync_service
+ global sync_manager
try:
- logger.info("Initializing MySQL sync service in background...")
+ logger.info("Initializing sync services in background...")
- # Run SyncService initialization in thread pool to avoid blocking event loop
- # SyncService.__init__() contains synchronous MySQL connection checks
+ # Run SyncServiceManager initialization in thread pool to avoid blocking event loop
+ # SyncServiceManager.__init__() contains synchronous data source connection checks
loop = asyncio.get_event_loop()
- sync_service = await loop.run_in_executor(None, SyncService)
+ sync_manager = await loop.run_in_executor(None, SyncServiceManager)
- # Start initial sync in background and wait for it to complete before starting auto sync
- # This ensures API starts immediately without blocking, but auto sync waits for initial sync
- async def run_sync():
- """Run sync_all in background without blocking"""
- try:
- logger.info("Starting initial data sync in background...")
- await sync_service.sync_all() # 全量同步 MySQL 数据到向量库
- logger.info("✓ Initial sync completed")
- return True
- except Exception as e:
- logger.error(f"Error during initial sync: {e}")
- return False
-
- # Create task for initial sync (don't await - allows API to start immediately)
- initial_sync_task = asyncio.create_task(run_sync())
-
- # Start auto sync if enabled
- # Auto sync service will wait for initial sync to complete before starting incremental syncs
- # 自动同步服务将在初始同步（全量同步）完成后开始增量同步
+ # Start all sync services in background
+ # This will start auto sync with recovery for each data source
if settings.AUTO_SYNC:
- logger.info(f"Starting auto sync service (interval: {settings.SYNC_INTERVAL}s)...")
- # Start auto sync in background - it will wait for initial sync to complete
- async def start_auto_sync_after_init():
- """Wait for initial sync to complete, then start auto sync with error recovery"""
- try:
- # Wait for initial sync to complete (no timeout - wait indefinitely)
- logger.info("Auto sync service waiting for initial sync to complete (no timeout, will wait until completion)...")
- try:
- initial_sync_success = await initial_sync_task
- if initial_sync_success:
- logger.info("✓ Initial sync completed successfully")
- else:
- logger.warning("Initial sync failed, but starting auto sync service anyway...")
- except Exception as sync_error:
- logger.error(f"Initial sync encountered an error: {sync_error}", exc_info=True)
- logger.warning("Starting auto sync service despite initial sync error...")
-
- # Start auto sync service with error recovery (it will skip initial sync since we already did it)
- await sync_service.start_auto_sync_with_recovery(skip_initial_sync=True)
- except Exception as e:
- logger.error(f"Error starting auto sync service: {e}", exc_info=True)
-
- # Start auto sync service in background (don't await - allows API to start immediately)
- asyncio.create_task(start_auto_sync_after_init())
+ logger.info(f"Starting auto sync services (interval: {settings.SYNC_INTERVAL}s)...")
+ # Start all sync services in background
+ await sync_manager.start_all_sync_services()
else:
- # If auto sync is disabled, just run initial sync in background
- # Don't await it to avoid blocking
- pass
+ logger.info("Auto sync is disabled, skipping auto sync service startup")
except Exception as e:
- logger.error(f"Failed to initialize sync service: {e}")
- logger.warning("API is still available, but MySQL sync is disabled. Please check MySQL connection.")
+ logger.error(f"Failed to initialize sync services: {e}")
+ logger.warning("API is still available, but sync services are disabled. Please check data source connections.")
# Don't raise - allow API to continue running
# Start sync service initialization in background (non-blocking)
@@ -180,11 +140,9 @@ async def lifespan(app: FastAPI):
await auto_sync_task
except asyncio.CancelledError:
pass
- if sync_service:
- if hasattr(sync_service, 'stop_auto_sync'):
- sync_service.stop_auto_sync()
- if hasattr(sync_service, 'close'):
- sync_service.close()
+ if sync_manager:
+ sync_manager.stop_all_sync_services()
+ sync_manager.close_all()
logger.info("RAG services shut down")
@@ -336,6 +294,7 @@ class SyncRequest(BaseModel):
"""Manual sync request model"""
full_sync: bool = Field(False, description="Whether to perform full sync")
force: bool = Field(False, description="Whether to force re-processing of all documents (even if they exist)")
+ source_name: Optional[str] = Field(None, description="Specific data source to sync (optional, syncs all if not provided)")
class HealthResponse(BaseModel):
@@ -1051,6 +1010,29 @@ async def retrieve(request: RetrieveRequest):
raise HTTPException(status_code=500, detail=f"Error retrieving documents: {str(e)}")
+@app.delete("/documents/source/{source_name}")
+async def delete_documents_by_source(source_name: str):
+ """
+ Delete all documents from a specific data source
+
+ Args:
+ source_name: Name of the data source to delete documents from
+
+ Returns:
+ Deletion status
+ """
+ if vector_store_manager is None:
+ raise HTTPException(status_code=503, detail="Vector store manager not initialized")
+
+ try:
+ # Delete documents by source name
+ vector_store_manager.delete_documents_by_source(source_name)
+ return {"status": "success", "message": f"Deleted all documents from {source_name}"}
+ except Exception as e:
+ logger.error(f"Error deleting documents by source: {e}")
+ raise HTTPException(status_code=500, detail=f"Error deleting documents: {str(e)}")
+
+
@app.post("/sync")
async def manual_sync(request: SyncRequest):
"""
@@ -1062,17 +1044,47 @@ async def manual_sync(request: SyncRequest):
Returns:
Sync status
"""
- if sync_service is None:
+ if sync_manager is None:
raise HTTPException(status_code=503, detail="Sync service not initialized")
try:
- if request.full_sync:
- await sync_service.sync_all(force=request.force)
- message = "Full sync completed" + (" (forced re-processing)" if request.force else "")
- return {"status": "success", "message": message}
+ if request.source_name:
+ # Sync specific data source
+ if hasattr(sync_manager, 'get_sync_service'): # Check if it's a SyncServiceManager
+ specific_sync_service = sync_manager.get_sync_service(request.source_name)
+ if not specific_sync_service:
+ raise HTTPException(status_code=404, detail=f"Sync service not found for source: {request.source_name}")
+
+ if request.full_sync:
+ await specific_sync_service.sync_all(force=request.force, is_manual=True)
+ message = f"Full sync completed for {request.source_name}" + (" (forced re-processing)" if request.force else "")
+ else:
+ await specific_sync_service.sync_incremental(is_manual=True)
+ message = f"Incremental sync completed for {request.source_name}"
+ else:
+ # Old SyncService that handles all sources
+ if request.full_sync:
+ await sync_manager.sync_all(force=request.force)
+ message = "Full sync completed for all sources" + (" (forced re-processing)" if request.force else "")
+ else:
+ await sync_manager.sync_incremental()
+ message = "Incremental sync completed for all sources"
else:
- await sync_service.sync_incremental()
- return {"status": "success", "message": "Incremental sync completed"}
+ # Sync all data sources
+ if hasattr(sync_manager, 'start_all_sync_services'):
+ # This is a SyncServiceManager, it starts auto sync services which do initial sync
+ await sync_manager.start_all_sync_services()
+ message = "Started synchronization for all data sources"
+ else:
+ # Old SyncService
+ if request.full_sync:
+ await sync_manager.sync_all(force=request.force)
+ message = "Full sync completed for all sources" + (" (forced re-processing)" if request.force else "")
+ else:
+ await sync_manager.sync_incremental()
+ message = "Incremental sync completed for all sources"
+
+ return {"status": "success", "message": message}
except Exception as e:
logger.error(f"Error during sync: {e}")
raise HTTPException(status_code=500, detail=f"Sync failed: {str(e)}")
@@ -1314,45 +1326,286 @@ async def get_folder_configs():
List of all configurations with their IDs and details
"""
try:
- config_dir = os.path.join(os.path.dirname(__file__), "..", "configs")
- if not os.path.exists(config_dir):
- os.makedirs(config_dir)
- return {"configurations": []}
+ # Load configs from SQLite database
+ DATA_DIR = Path(__file__).parent.parent / "data"
+ DATA_DIR.mkdir(parents=True, exist_ok=True)
+ DB_PATH = DATA_DIR / "sessions.db"
- configurations = []
- config_files = [f for f in os.listdir(config_dir) if f.endswith(".json")]
+ conn = sqlite3.connect(DB_PATH)
+ cursor = conn.cursor()
- for config_file in config_files:
- config_path = os.path.join(config_dir, config_file)
- with open(config_path, "r", encoding="utf-8") as f:
- config_data = json.load(f)
+ # 查询所有数据源配置
+ try:
+ cursor.execute('SELECT name, config, update_at FROM data_sources')
+ rows = cursor.fetchall()
- # Handle both single config and array
- if isinstance(config_data, list):
- for idx, config in enumerate(config_data):
+ configurations = []
+ for row in rows:
+ name, config_json, update_at = row
+ try:
+ config_data = json.loads(config_json)
configurations.append({
- "id": f"{os.path.splitext(config_file)[0]}-{idx}",
- "type": config.get("type", "unknown"),
- "config": config
+ "id": name,
+ "type": config_data.get("type", "unknown"),
+ "config": config_data,
+ "update_at": update_at
})
- else:
- configurations.append({
- "id": os.path.splitext(config_file)[0],
- "type": config_data.get("type", "unknown"),
- "config": config_data
- })
+ except Exception as e:
+ logger.error(f"Error parsing config from database: {name}, error: {e}")
+ except sqlite3.OperationalError as e:
+ # 表不存在的情况，返回空列表
+ logger.warning(f"SQLite table error: {e}. Returning empty config list.")
+ configurations = []
+
+ # 关闭数据库连接
+ conn.close()
return {"configurations": configurations}
except Exception as e:
logger.error(f"Error getting folder configs: {e}")
- raise HTTPException(status_code=500, detail=f"Error getting configurations: {str(e)}")
+ # 返回空列表而不是错误响应
+ return {"configurations": []}
+
+
+@app.post("/folder-configs")
+async def create_config(config: Dict[str, Any]):
+ """
+ Create a new configuration
+
+ Args:
+ config: Configuration data for the new config
+
+ Returns:
+ Created configuration with ID
+ """
+ conn = None
+ try:
+ # Ensure config has required fields
+ if "type" not in config:
+ raise HTTPException(status_code=400, detail="Config type is required")
+
+ config_type = config["type"]
+
+ # Check for unique identifier based on config type
+ # 必须有数据源唯一标识才能创建配置
+ unique_id_parts = [config_type]
+
+ if config_type == "database":
+ # 数据库配置需要：主机、端口、数据库名、表名
+ 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="数据库配置必须包含主机、端口、数据库名和表名")
+ unique_id_parts.extend([
+ config["mysql_host"].lower(),
+ str(config["mysql_port"]),
+ config["database"].lower(),
+ config["table_name"].lower()
+ ])
+ elif config_type == "local_folder":
+ # 本地文件夹配置需要：文件夹路径
+ if not config.get("folder_path"):
+ raise HTTPException(status_code=400, detail="本地文件夹配置必须包含文件夹路径")
+ # 替换路径中的特殊字符为下划线
+ folder_path = config["folder_path"].lower().replace("/", "_").replace(":", "_").replace("\\", "_")
+ # 去除开头的下划线
+ if folder_path.startswith("_"):
+ folder_path = folder_path[1:]
+ unique_id_parts.append(folder_path)
+ elif config_type == "remote_folder":
+ # 远程文件夹配置需要：主机、文件夹路径
+ if not config.get("host") or not config.get("folder_path"):
+ raise HTTPException(status_code=400, detail="远程文件夹配置必须包含主机和文件夹路径")
+ # 替换路径中的特殊字符为下划线
+ folder_path = config["folder_path"].lower().replace("/", "_").replace(":", "_").replace("\\", "_")
+ # 去除开头的下划线
+ if folder_path.startswith("_"):
+ folder_path = folder_path[1:]
+ unique_id_parts.extend([
+ config["host"].lower(),
+ folder_path
+ ])
+ else:
+ raise HTTPException(status_code=400, detail=f"不支持的配置类型: {config_type}")
+
+ # Generate config ID based on unique identifier
+ config_id = "_".join(unique_id_parts)
+
+ # Save config to SQLite database
+ conn, cursor = get_db_connection()
+
+ # 检查是否存在同源配置
+ try:
+ # 先尝试查询现有配置
+ cursor.execute('SELECT name, config FROM data_sources')
+ existing_configs = cursor.fetchall()
+
+ for existing_name, existing_config_json in existing_configs:
+ existing_config_data = json.loads(existing_config_json)
+
+ # 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
+ existing_config_data.get('mysql_port') == config.get('mysql_port') and
+ existing_config_data.get('database') == config.get('database')):
+ raise HTTPException(
+ status_code=409,
+ detail=f"已存在相同源的数据库配置。如需调整，请点击配置列表中的配置并修改配置内容。"
+ )
+ elif config_type == 'local_folder':
+ # For local folder configs, same source means same folder path
+ if existing_config_data.get('folder_path') == config.get('folder_path'):
+ raise HTTPException(
+ status_code=409,
+ detail=f"已存在相同路径的本地文件夹配置。如需调整，请点击配置列表中的配置并修改配置内容。"
+ )
+ elif config_type == 'remote_folder':
+ # For remote folder configs, same source means same host and folder path
+ if (existing_config_data.get('host') == config.get('host') and
+ existing_config_data.get('folder_path') == config.get('folder_path')):
+ raise HTTPException(
+ status_code=409,
+ detail=f"已存在相同服务器和路径的远程文件夹配置。如需调整，请点击配置列表中的配置并修改配置内容。"
+ )
+ except sqlite3.OperationalError as e:
+ # 表不存在的情况，会在后面创建表
+ logger.warning(f"SQLite table error: {e}. This is expected if the table doesn't exist yet.")
+
+
+
+ # Check if a config with the same name already exists
+ cursor.execute('SELECT name FROM data_sources WHERE name = ?', (config_id,))
+ existing_config = cursor.fetchone()
+
+ if existing_config:
+ raise HTTPException(
+ status_code=409,
+ detail=f"配置 '{config['name']}' 已存在。如需调整，请点击配置列表中的配置并修改配置内容。"
+ )
+
+ # Insert new config into data_sources table
+ config_json = json.dumps(config, ensure_ascii=False, indent=2)
+
+ cursor.execute(
+ 'INSERT INTO data_sources (name, config) VALUES (?, ?)',
+ (config_id, config_json)
+ )
+
+ conn.commit()
+
+ # Create sync service for this new data source
+ global sync_manager
+ if sync_manager is not None:
+ # Create appropriate data source config object
+ from config import BaseDataSourceConfig, DatabaseDataSourceConfig, LocalFolderDataSourceConfig, RemoteFolderDataSourceConfig
+
+ if config_type == "database":
+ source_config = DatabaseDataSourceConfig(
+ name=config_id,
+ database=config.get("database"),
+ table_name=config.get("table_name"),
+ id_column=config.get("id_column"),
+ content_column=config.get("content_column"),
+ file_column=config.get("file_column"),
+ title_column=config.get("title_column"),
+ 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"),
+ file_source_type=config.get("file_source_type"),
+ file_system_base_path=config.get("file_system_base_path"),
+ scp_host=config.get("scp_host"),
+ scp_port=config.get("scp_port"),
+ scp_username=config.get("scp_username"),
+ scp_password=config.get("scp_password")
+ )
+ elif config_type == "local_folder":
+ source_config = LocalFolderDataSourceConfig(
+ name=config_id,
+ folder_path=config.get("folder_path"),
+ host=config.get("host"),
+ port=config.get("port"),
+ username=config.get("username"),
+ password=config.get("password"),
+ recursive=config.get("recursive", True),
+ ignore_patterns=config.get("ignore_patterns")
+ )
+ elif config_type == "remote_folder":
+ source_config = RemoteFolderDataSourceConfig(
+ name=config_id,
+ folder_path=config.get("folder_path"),
+ host=config.get("host"),
+ port=config.get("port"),
+ username=config.get("username"),
+ password=config.get("password"),
+ recursive=config.get("recursive", True),
+ ignore_patterns=config.get("ignore_patterns")
+ )
+ else:
+ logger.warning(f"Unknown config type: {config_type}")
+ # Create a base config as fallback
+ source_config = BaseDataSourceConfig(name=config_id, type=config_type)
+
+ # Create the sync service
+ sync_manager.create_or_update_sync_service(source_config)
+
+ # Return the created config
+ return {
+ "id": config_id,
+ "type": config_type,
+ "config": config
+ }
+
+ except HTTPException:
+ raise
+ except Exception as e:
+ logger.error(f"Error creating config: {e}")
+ raise HTTPException(status_code=500, detail=f"Error creating configuration: {str(e)}")
+ finally:
+ if conn:
+ conn.close()
+
+
+def get_db_connection():
+ """
+ Get a SQLite database connection with data_sources table initialized
+
+ Returns:
+ tuple: (connection, cursor)
+ """
+ DATA_DIR = Path(__file__).parent.parent / "data"
+ DATA_DIR.mkdir(parents=True, exist_ok=True)
+ DB_PATH = DATA_DIR / "sessions.db"
+
+ conn = sqlite3.connect(DB_PATH)
+ cursor = conn.cursor()
+
+ # 检查并创建data_sources表（如果不存在）
+ try:
+ cursor.execute('''
+ CREATE TABLE IF NOT EXISTS data_sources (
+ name TEXT PRIMARY KEY,
+ config TEXT NOT NULL,
+ update_at TEXT NULL
+ )
+ ''')
+ conn.commit()
+ except Exception as e:
+ logger.error(f"Error creating data_sources table: {e}")
+ conn.close()
+ raise HTTPException(status_code=500, detail=f"Error creating configuration table: {str(e)}")
+
+ return conn, cursor
@app.post("/folder-configs/remote")
async def create_remote_folder_config(config: Dict[str, Any]):
"""
- Create a new remote folder configuration
+ Create a new remote folder configuration (deprecated, use /folder-configs instead)
Args:
config: Configuration data for the new remote folder
@@ -1360,34 +1613,12 @@ async def create_remote_folder_config(config: Dict[str, Any]):
Returns:
Created configuration with ID
"""
- try:
- config_dir = os.path.join(os.path.dirname(__file__), "..", "configs")
- if not os.path.exists(config_dir):
- os.makedirs(config_dir)
-
- # Generate unique config ID
- config_id = f"remote_folder_{int(time.time())}"
- config_file = f"{config_id}.json"
- config_path = os.path.join(config_dir, config_file)
-
- # Ensure config has required fields
- if "type" not in config:
- config["type"] = "remote_folder"
-
- # Write config to file
- with open(config_path, "w", encoding="utf-8") as f:
- json.dump(config, f, ensure_ascii=False, indent=2)
-
- # Return the created config
- return {
- "id": config_id,
- "type": config["type"],
- "config": config
- }
-
- except Exception as e:
- logger.error(f"Error creating remote folder config: {e}")
- raise HTTPException(status_code=500, detail=f"Error creating configuration: {str(e)}")
+ # Set type to remote_folder if not provided
+ if "type" not in config:
+ config["type"] = "remote_folder"
+
+ # Call the generic create_config function
+ return await create_config(config)
@app.put("/folder-configs/{config_id}")
@@ -1402,26 +1633,140 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]):
Returns:
Updated configuration
"""
+ conn = None
try:
- config_dir = os.path.join(os.path.dirname(__file__), "..", "configs")
- config_file = f"{config_id}.json"
- config_path = os.path.join(config_dir, config_file)
+ conn, cursor = get_db_connection()
- if not os.path.exists(config_path):
+ # Check if config exists
+ cursor.execute('SELECT * FROM data_sources WHERE name = ?', (config_id,))
+ existing_config = cursor.fetchone()
+
+ if not existing_config:
raise HTTPException(status_code=404, detail=f"Configuration with ID '{config_id}' not found")
# Ensure config has required fields
if "type" not in config:
- config["type"] = "remote_folder"
+ raise HTTPException(status_code=400, detail="Configuration type is required")
- # Update config file
- with open(config_path, "w", encoding="utf-8") as f:
- json.dump(config, f, ensure_ascii=False, indent=2)
+ # Generate new config ID based on the complete configuration information
+ config_type = config["type"]
+ new_config_id = None
- # Return the updated config
+ # 根据不同类型的配置生成有意义的ID
+ if config_type == "database":
+ # 数据库配置：使用数据库名和表名生成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()}"
+ elif config_type == "local_folder":
+ # 本地文件夹配置：使用文件夹路径生成ID
+ if not config.get("folder_path"):
+ raise HTTPException(status_code=400, detail="Folder path is required for local folder configuration")
+ # 替换路径中的特殊字符为下划线
+ folder_path = config["folder_path"].lower().replace("/", "_").replace(":", "_").replace("\\", "_")
+ # 去除开头的下划线
+ if folder_path.startswith("_"):
+ folder_path = folder_path[1:]
+ new_config_id = f"{config_type}_{folder_path}"
+ elif config_type == "remote_folder":
+ # 远程文件夹配置：使用主机和文件夹路径生成ID
+ # 端口不是唯一标识的必要元素，只用于连接配置
+ if not config.get("host"):
+ raise HTTPException(status_code=400, detail="Host is required for remote folder configuration")
+ if not config.get("folder_path"):
+ raise HTTPException(status_code=400, detail="Folder path is required for remote folder configuration")
+ # 替换路径中的特殊字符为下划线
+ folder_path = config["folder_path"].lower().replace("/", "_").replace(":", "_").replace("\\", "_")
+ # 去除开头的下划线
+ if folder_path.startswith("_"):
+ folder_path = folder_path[1:]
+ new_config_id = f"{config_type}_{config['host'].lower()}_{folder_path}"
+
+ # 如果无法生成新的有意义的ID，保留原来的ID
+ if not new_config_id:
+ new_config_id = config_id
+
+ # Update config in data_sources table
+ # Since name is primary key, we need to use INSERT OR REPLACE
+ config_json = json.dumps(config, ensure_ascii=False, indent=2)
+
+ # First, delete the old config if new ID is different
+ if new_config_id != config_id:
+ cursor.execute('DELETE FROM data_sources WHERE name = ?', (config_id,))
+
+ # Then insert the updated config with new ID, update_at value to null
+ cursor.execute(
+ 'INSERT OR REPLACE INTO data_sources (name, config, update_at) VALUES (?, ?, ?)',
+ (new_config_id, config_json, None)
+ )
+
+ conn.commit()
+
+ # Update the sync service for this data source
+ global sync_manager
+ if sync_manager is not None:
+ # Create appropriate data source config object
+ from config import BaseDataSourceConfig, DatabaseDataSourceConfig, LocalFolderDataSourceConfig, RemoteFolderDataSourceConfig
+
+ if config_type == "database":
+ source_config = DatabaseDataSourceConfig(
+ name=new_config_id,
+ database=config.get("database"),
+ table_name=config.get("table_name"),
+ id_column=config.get("id_column"),
+ content_column=config.get("content_column"),
+ file_column=config.get("file_column"),
+ title_column=config.get("title_column"),
+ 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"),
+ file_source_type=config.get("file_source_type"),
+ file_system_base_path=config.get("file_system_base_path"),
+ scp_host=config.get("scp_host"),
+ scp_port=config.get("scp_port"),
+ scp_username=config.get("scp_username"),
+ scp_password=config.get("scp_password")
+ )
+ elif config_type == "local_folder":
+ source_config = LocalFolderDataSourceConfig(
+ name=new_config_id,
+ folder_path=config.get("folder_path"),
+ host=config.get("host"),
+ port=config.get("port"),
+ username=config.get("username"),
+ password=config.get("password"),
+ recursive=config.get("recursive", True),
+ ignore_patterns=config.get("ignore_patterns")
+ )
+ elif config_type == "remote_folder":
+ source_config = RemoteFolderDataSourceConfig(
+ name=new_config_id,
+ folder_path=config.get("folder_path"),
+ host=config.get("host"),
+ port=config.get("port"),
+ username=config.get("username"),
+ password=config.get("password"),
+ recursive=config.get("recursive", True),
+ ignore_patterns=config.get("ignore_patterns")
+ )
+ else:
+ logger.warning(f"Unknown config type: {config_type}")
+ # Create a base config as fallback
+ source_config = BaseDataSourceConfig(name=new_config_id, type=config_type)
+
+ # Update the sync service
+ sync_manager.create_or_update_sync_service(source_config)
+
+ # Return the updated config with new ID
return {
- "id": config_id,
- "type": config["type"],
+ "id": new_config_id,
+ "type": config_type,
"config": config
}
@@ -1430,6 +1775,9 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]):
except Exception as e:
logger.error(f"Error updating folder config: {e}")
raise HTTPException(status_code=500, detail=f"Error updating configuration: {str(e)}")
+ finally:
+ if conn:
+ conn.close()
@app.delete("/folder-configs/{config_id}")
@@ -1443,16 +1791,26 @@ async def delete_folder_config(config_id: str):
Returns:
Success message
"""
+ conn = None
try:
- config_dir = os.path.join(os.path.dirname(__file__), "..", "configs")
- config_file = f"{config_id}.json"
- config_path = os.path.join(config_dir, config_file)
+ conn, cursor = get_db_connection()
- if not os.path.exists(config_path):
+ # Check if config exists
+ cursor.execute('SELECT * FROM data_sources WHERE name = ?', (config_id,))
+ existing_config = cursor.fetchone()
+
+ if not existing_config:
raise HTTPException(status_code=404, detail=f"Configuration with ID '{config_id}' not found")
- # Delete config file
- os.remove(config_path)
+ # Delete config from data_sources table
+ cursor.execute('DELETE FROM data_sources WHERE name = ?', (config_id,))
+
+ conn.commit()
+
+ # Remove the sync service for this data source
+ global sync_manager
+ if sync_manager is not None:
+ sync_manager.remove_sync_service(config_id)
return {"status": "success", "message": f"Configuration '{config_id}' deleted successfully"}
@@ -1461,6 +1819,173 @@ async def delete_folder_config(config_id: str):
except Exception as e:
logger.error(f"Error deleting folder config: {e}")
raise HTTPException(status_code=500, detail=f"Error deleting configuration: {str(e)}")
+ finally:
+ if conn:
+ conn.close()
+
+
+# Database schema exploration endpoints
+from pydantic import BaseModel
+import mysql.connector
+
+
+
+class DatabaseConnectionParams(BaseModel):
+ """Database connection parameters"""
+ host: str = Field(..., description="Database host")
+ port: int = Field(default=3306, description="Database port")
+ username: str = Field(..., description="Database username")
+ password: str = Field(..., description="Database password")
+
+class DatabaseParams(BaseModel):
+ """Database parameters"""
+ host: str = Field(..., description="Database host")
+ port: int = Field(default=3306, description="Database port")
+ username: str = Field(..., description="Database username")
+ password: str = Field(..., description="Database password")
+ database: str = Field(..., description="Database name")
+
+class TableParams(BaseModel):
+ """Table parameters"""
+ host: str = Field(..., description="Database host")
+ port: int = Field(default=3306, description="Database port")
+ username: str = Field(..., description="Database username")
+ password: str = Field(..., description="Database password")
+ database: str = Field(..., description="Database name")
+ table_name: str = Field(..., description="Table name")
+
+
+
+
+@app.post("/database/databases")
+async def get_databases(params: DatabaseConnectionParams):
+ """
+ Get list of databases from MySQL server
+
+ Args:
+ params: Database connection parameters
+
+ Returns:
+ 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()
+
+ return {"databases": databases}
+
+ except mysql.connector.Error 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")
+async def get_tables(params: DatabaseParams):
+ """
+ Get list of tables from specified database
+
+ Args:
+ params: Database parameters including database name
+
+ Returns:
+ 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()
+
+ return {"tables": tables}
+
+ except mysql.connector.Error 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")
+async def get_table_structure(params: TableParams):
+ """
+ Get structure of specified table
+
+ Args:
+ params: Table parameters including table name
+
+ Returns:
+ 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()
+
+ return {"columns": columns}
+
+ except mysql.connector.Error 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 7112fc7..d4a286c 100644
--- a/config.py
+++ b/config.py
@@ -34,7 +34,6 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig):
mysql_port: Optional[int] = None,
mysql_user: Optional[str] = None,
mysql_password: Optional[str] = None,
- mysql_charset: Optional[str] = None,
# 文件源配置
file_source_type: Optional[str] = None, # 可选值: "api", "filesystem", "scp"
file_system_base_path: Optional[str] = None, # 文件系统基础路径
@@ -67,7 +66,6 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig):
self.mysql_port = mysql_port
self.mysql_user = mysql_user
self.mysql_password = mysql_password
- self.mysql_charset = mysql_charset
# 文件源配置
self.file_source_type = file_source_type
@@ -78,7 +76,6 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig):
self.scp_port = scp_port
self.scp_username = scp_username
self.scp_password = scp_password
- self.scp_key_path = scp_key_path
class LocalFolderDataSourceConfig(BaseDataSourceConfig):
@@ -87,11 +84,19 @@ class LocalFolderDataSourceConfig(BaseDataSourceConfig):
self,
name: str,
folder_path: str,
+ host: str = "localhost",
+ port: int = 22,
+ username: Optional[str] = None,
+ password: Optional[str] = None,
recursive: bool = True,
ignore_patterns: Optional[List[str]] = None
):
super().__init__(name, "local_folder")
self.folder_path = folder_path # 本地文件夹路径
+ self.host = host # 主机地址
+ self.port = port # 主机端口
+ self.username = username # 主机用户名
+ self.password = password # 主机密码（可选）
self.recursive = recursive # 是否递归遍历子文件夹
self.ignore_patterns = ignore_patterns # 忽略的文件模式列表
@@ -106,7 +111,6 @@ class RemoteFolderDataSourceConfig(BaseDataSourceConfig):
port: int,
username: str,
password: Optional[str] = None,
- key_path: Optional[str] = None,
recursive: bool = True,
ignore_patterns: Optional[List[str]] = None
):
@@ -116,7 +120,6 @@ class RemoteFolderDataSourceConfig(BaseDataSourceConfig):
self.port = port # 远程主机端口
self.username = username # 远程主机用户名
self.password = password # 远程主机密码（可选）
- self.key_path = key_path # SSH密钥路径（可选）
self.recursive = recursive # 是否递归遍历子文件夹
self.ignore_patterns = ignore_patterns # 忽略的文件模式列表
@@ -140,21 +143,6 @@ class Settings(BaseSettings):
# File Upload Settings
MAX_UPLOAD_SIZE_MB: int = 5 # Maximum file upload size in MB (default: 5MB)
-
-
- # Multi-database configuration (JSON string or file path) - legacy
- # Format: JSON array with database configs (same MySQL server, different databases)
- # Each config only needs: name, database, table_name, id_column, content_column, etc.
- # Connection info (host, port, user, password) will use MYSQL_* settings above
- # Set this in .env file: MYSQL_DATABASES_CONFIG=./databases_config.json
- MYSQL_DATABASES_CONFIG: Optional[str] = None
-
- # Multi-data source configuration (JSON string or file path) - new
- # Format: JSON array with data source configs of different types
- # Each config needs: name, type (database, local_folder, remote_folder), and type-specific settings
- # Set this in .env file: DATA_SOURCES_CONFIG=./data_sources_config.json
- DATA_SOURCES_CONFIG: Optional[str] = None
-
# ChromaDB Settings
# Use HttpClient mode if CHROMA_SERVER_HOST is set, otherwise use PersistentClient
# For Docker deployment, set CHROMA_SERVER_HOST=localhost in .env
@@ -213,175 +201,40 @@ class Settings(BaseSettings):
Get list of data source configurations
Returns:
- List of BaseDataSourceConfig objects
+ List of BaseDataSourceConfig objects. Empty list if no configurations found.
"""
configs = []
try:
- # Check if data sources config is provided
- config_source = self.DATA_SOURCES_CONFIG
+ # Load configs from SQLite database only
+ import sqlite3
+ from pathlib import Path
- if config_source:
- # Try to parse as JSON string
- if config_source.strip().startswith('[') or config_source.strip().startswith('{'):
- data_sources = json.loads(config_source)
- # Handle both array and single object
- if isinstance(data_sources, dict):
- data_sources = [data_sources]
+ DATA_DIR = Path(__file__).parent / "data"
+ DATA_DIR.mkdir(parents=True, exist_ok=True)
+ DB_PATH = DATA_DIR / "sessions.db"
+
+ try:
+ conn = sqlite3.connect(DB_PATH)
+ cursor = conn.cursor()
+
+ # 查询所有数据源配置
+ try:
+ cursor.execute('SELECT name, config FROM data_sources')
+ rows = cursor.fetchall()
- # Create data source objects based on type
- for ds_config in data_sources:
- source_type = ds_config.get('type', 'database')
-
- if source_type == 'database':
- # Create database data source
- configs.append(DatabaseDataSourceConfig(
- name=ds_config.get('name', ds_config.get('database', 'unknown')),
- database=ds_config['database'], # Required field
- table_name=ds_config.get('table_name', 'documents'), # Default to 'documents'
- 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),
- title_column=ds_config.get('title_column', 'title'), # Default to 'title'
- 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),
- mysql_charset=ds_config.get('mysql_charset', 'utf8mb4'), # Default to 'utf8mb4'
- # 文件源配置
- file_source_type=ds_config.get('file_source_type', None),
- file_system_base_path=ds_config.get('file_system_base_path', None),
- # SCP配置
- scp_host=ds_config.get('scp_host', None),
- scp_port=ds_config.get('scp_port', 22),
- scp_username=ds_config.get('scp_username', None),
- scp_password=ds_config.get('scp_password', None),
- scp_key_path=ds_config.get('scp_key_path', None)
- ))
- elif source_type == 'local_folder':
- # Create local folder data source
- configs.append(LocalFolderDataSourceConfig(
- name=ds_config.get('name', 'local_folder'),
- folder_path=ds_config.get('folder_path', '.'),
- recursive=ds_config.get('recursive', True),
- ignore_patterns=ds_config.get('ignore_patterns', None)
- ))
- elif source_type == 'remote_folder':
- # Create remote folder data source
- configs.append(RemoteFolderDataSourceConfig(
- name=ds_config.get('name', 'remote_folder'),
- folder_path=ds_config.get('folder_path', '.'),
- host=ds_config.get('host', ''),
- port=ds_config.get('port', 22),
- username=ds_config.get('username', ''),
- password=ds_config.get('password', None),
- key_path=ds_config.get('key_path', None),
- recursive=ds_config.get('recursive', True),
- ignore_patterns=ds_config.get('ignore_patterns', None)
- ))
- else:
- from loguru import logger
- logger.warning(f"Unknown data source type: {source_type}, skipping")
- else:
- # Load all config files from configs directory
- config_dir = os.path.join(os.path.dirname(__file__), "configs")
- if os.path.exists(config_dir):
- import glob
- for config_file in glob.glob(os.path.join(config_dir, "*.json")):
+ for row in rows:
+ name, config_json = row
try:
- with open(config_file, 'r', encoding='utf-8') as f:
- data_sources = json.load(f)
-
- # Handle both array and single object
- if isinstance(data_sources, dict):
- data_sources = [data_sources]
+ ds_config = json.loads(config_json)
# Create data source objects based on type
- for ds_config in data_sources:
- source_type = ds_config.get('type', 'database')
-
- if source_type == 'database':
- # Create database data source
- configs.append(DatabaseDataSourceConfig(
- name=ds_config.get('name', ds_config.get('database', 'unknown')),
- database=ds_config['database'], # Required field
- table_name=ds_config.get('table_name', 'documents'), # Default to 'documents'
- 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),
- title_column=ds_config.get('title_column', 'title'), # Default to 'title'
- 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),
- mysql_charset=ds_config.get('mysql_charset', 'utf8mb4'), # Default to 'utf8mb4'
- # 文件源配置
- file_source_type=ds_config.get('file_source_type', None),
- file_system_base_path=ds_config.get('file_system_base_path', None),
- # SCP配置
- scp_host=ds_config.get('scp_host', None),
- scp_port=ds_config.get('scp_port', 22),
- scp_username=ds_config.get('scp_username', None),
- scp_password=ds_config.get('scp_password', None),
- scp_key_path=ds_config.get('scp_key_path', None)
- ))
- elif source_type == 'local_folder':
- # Create local folder data source
- configs.append(LocalFolderDataSourceConfig(
- name=ds_config.get('name', 'local_folder'),
- folder_path=ds_config.get('folder_path', '.'),
- recursive=ds_config.get('recursive', True),
- ignore_patterns=ds_config.get('ignore_patterns', None)
- ))
- elif source_type == 'remote_folder':
- # Create remote folder data source
- configs.append(RemoteFolderDataSourceConfig(
- name=ds_config.get('name', 'remote_folder'),
- folder_path=ds_config.get('folder_path', '.'),
- host=ds_config.get('host', ''),
- port=ds_config.get('port', 22),
- username=ds_config.get('username', ''),
- password=ds_config.get('password', None),
- key_path=ds_config.get('key_path', None),
- recursive=ds_config.get('recursive', True),
- ignore_patterns=ds_config.get('ignore_patterns', None)
- ))
- else:
- from loguru import logger
- logger.warning(f"Unknown data source type: {source_type}, skipping")
-
- from loguru import logger
- logger.info(f"Loaded config file: {config_file}")
- except Exception as e:
- from loguru import logger
- logger.error(f"Error loading config file {config_file}: {e}")
- else:
- # Fallback to default database_config.json if configs directory doesn't exist
- default_config_path = os.path.join(os.path.dirname(__file__), "database_config.json")
- if os.path.exists(default_config_path):
- with open(default_config_path, 'r', encoding='utf-8') as f:
- data_sources = json.load(f)
-
- # Handle both array and single object
- if isinstance(data_sources, dict):
- data_sources = [data_sources]
-
- # Create data source objects based on type
- for ds_config in data_sources:
source_type = ds_config.get('type', 'database')
if source_type == 'database':
# Create database data source
configs.append(DatabaseDataSourceConfig(
- name=ds_config.get('name', ds_config.get('database', 'unknown')),
+ name=name, # 使用数据库表中的name列
database=ds_config['database'], # Required field
table_name=ds_config.get('table_name', 'documents'), # Default to 'documents'
id_column=ds_config.get('id_column', 'id'), # Default to 'id'
@@ -396,7 +249,6 @@ class Settings(BaseSettings):
mysql_port=ds_config.get('mysql_port', None),
mysql_user=ds_config.get('mysql_user', None),
mysql_password=ds_config.get('mysql_password', None),
- mysql_charset=ds_config.get('mysql_charset', 'utf8mb4'), # Default to 'utf8mb4'
# 文件源配置
file_source_type=ds_config.get('file_source_type', None),
file_system_base_path=ds_config.get('file_system_base_path', None),
@@ -404,47 +256,63 @@ class Settings(BaseSettings):
scp_host=ds_config.get('scp_host', None),
scp_port=ds_config.get('scp_port', 22),
scp_username=ds_config.get('scp_username', None),
- scp_password=ds_config.get('scp_password', None),
- scp_key_path=ds_config.get('scp_key_path', None)
+ scp_password=ds_config.get('scp_password', None)
))
elif source_type == 'local_folder':
# Create local folder data source
configs.append(LocalFolderDataSourceConfig(
- name=ds_config.get('name', 'local_folder'),
- folder_path=ds_config.get('folder_path', '.'),
- recursive=ds_config.get('recursive', True),
- ignore_patterns=ds_config.get('ignore_patterns', None)
- ))
+ name=name, # 使用数据库表中的name列
+ folder_path=ds_config.get('folder_path', '.'),
+ host=ds_config.get('host', 'localhost'),
+ port=ds_config.get('port', 22),
+ username=ds_config.get('username', None),
+ password=ds_config.get('password', None),
+ recursive=ds_config.get('recursive', True),
+ ignore_patterns=ds_config.get('ignore_patterns', None)
+ ))
elif source_type == 'remote_folder':
# Create remote folder data source
configs.append(RemoteFolderDataSourceConfig(
- name=ds_config.get('name', 'remote_folder'),
+ name=name, # 使用数据库表中的name列
folder_path=ds_config.get('folder_path', '.'),
host=ds_config.get('host', ''),
port=ds_config.get('port', 22),
username=ds_config.get('username', ''),
password=ds_config.get('password', None),
- key_path=ds_config.get('key_path', None),
recursive=ds_config.get('recursive', True),
ignore_patterns=ds_config.get('ignore_patterns', None)
))
else:
from loguru import logger
logger.warning(f"Unknown data source type: {source_type}, skipping")
- else:
- raise FileNotFoundError(f"Default config file not found: {default_config_path}")
-
+
+ from loguru import logger
+ logger.info(f"Loaded config from database: {name}")
+ except Exception as e:
+ from loguru import logger
+ logger.error(f"Error parsing config from database: {name}, error: {e}")
+ except sqlite3.OperationalError as e:
+ # 表不存在的情况，返回空列表
+ from loguru import logger
+ logger.warning(f"SQLite table error: {e}. Returning empty config list.")
+ except Exception as e:
+ # 其他数据库错误，返回空列表
+ from loguru import logger
+ logger.error(f"Error querying data sources from database: {e}")
+
+ # 关闭数据库连接
+ conn.close()
+
+ except Exception as e:
+ # 数据库连接失败，返回空列表
+ from loguru import logger
+ logger.error(f"Error connecting to SQLite database: {e}")
except Exception as e:
+ # 任何其他错误，返回空列表
from loguru import logger
- logger.error(f"Error parsing data sources config: {e}")
- raise ValueError(f"Failed to load data sources configuration: {e}") from e
-
- # Ensure we have at least one data source
- if not configs:
- from loguru import logger
- logger.error("No valid data sources found in configuration")
- raise ValueError("No valid data sources found in configuration")
+ logger.error(f"Error in get_data_sources: {e}")
+ # 返回配置列表，即使为空
return configs
def get_database_configs(self) -> List[DatabaseDataSourceConfig]:
diff --git a/configs/host_folder.json b/configs/host_folder.json
deleted file mode 100644
index cd5f756..0000000
--- a/configs/host_folder.json
+++ /dev/null
@@ -1,12 +0,0 @@
-{
- "type": "remote_folder",
- "name": "host_folder",
- "folder_path": "/home/zxh/programs/RAG/test_files",
- "host": "localhost",
- "port": 22,
- "username": "zxh",
- "password": "",
- "key_path": "/home/zxh/.ssh/id_rsa",
- "recursive": true,
- "ignore_patterns": [".git", ".DS_Store"]
-}
\ No newline at end of file
diff --git a/database/__init__.py b/database/__init__.py
deleted file mode 100644
index aed046d..0000000
--- a/database/__init__.py
+++ /dev/null
@@ -1,7 +0,0 @@
-"""
-Database module for RAG system
-"""
-from .sync import MySQLSync
-
-__all__ = ["MySQLSync"]
-
diff --git a/database/sync.py b/database/sync.py
deleted file mode 100644
index 9abbc30..0000000
--- a/database/sync.py
+++ /dev/null
@@ -1,907 +0,0 @@
-"""
-MySQL to ChromaDB synchronization module
-"""
-import urllib
-import pymysql
-import requests
-from typing import List, Dict, Optional, Tuple
-from datetime import datetime
-from loguru import logger
-from config import DatabaseDataSourceConfig as DatabaseConfig, settings
-from rag.file_parser import FileParser
-
-
-class MySQLSync:
- """Handle synchronization between MySQL and ChromaDB"""
-
- def __init__(self, db_config: DatabaseConfig, connection=None):
- """
- Initialize MySQL sync with database configuration
-
- Args:
- db_config: DatabaseConfig object containing database table info
- connection: Optional existing MySQL connection (for multi-database scenario)
- """
- self.file_parser = FileParser()
- self.db_config = db_config
- self.connection = connection
- if self.connection is None:
- self._connect()
- else:
- # Use existing connection, just switch database
- self._switch_database()
-
- def _connect(self):
- """Establish MySQL connection using config-specific connection info"""
- try:
- # Use database-specific connection info only
- if not all([self.db_config.mysql_host, self.db_config.mysql_port, self.db_config.mysql_user]):
- missing_fields = []
- if not self.db_config.mysql_host:
- missing_fields.append("mysql_host")
- if not self.db_config.mysql_port:
- missing_fields.append("mysql_port")
- if not self.db_config.mysql_user:
- missing_fields.append("mysql_user")
- raise ValueError(f"Missing required database connection fields for {self.db_config.name}: {', '.join(missing_fields)}")
-
- host = self.db_config.mysql_host
- port = self.db_config.mysql_port
- user = self.db_config.mysql_user
- password = self.db_config.mysql_password or "" # Password can be empty
- charset = self.db_config.mysql_charset or "utf8mb4" # Charset has a reasonable default
-
- self.connection = pymysql.connect(
- host=host,
- port=port,
- user=user,
- password=password,
- database=self.db_config.database,
- charset=charset,
- cursorclass=pymysql.cursors.DictCursor
- )
- connection_info = f"{host}:{port}/{self.db_config.database}"
- logger.info(f"Connected to MySQL database: {connection_info} (name: {self.db_config.name})")
- except Exception as e:
- logger.error(f"Failed to connect to MySQL database {self.db_config.name}: {e}")
- raise
-
- @staticmethod
- def check_mysql_connection(host="localhost", port=3306, user="root", password="", charset="utf8mb4") -> Tuple[bool, str]:
- """
- Check if MySQL server is accessible and connection can be established
-
- Args:
- host: MySQL host
- port: MySQL port
- user: MySQL username
- password: MySQL password
- charset: MySQL charset
-
- Returns:
- Tuple of (is_connected: bool, error_message: str)
- If connected, error_message will be empty string
- """
- temp_connection = None
- try:
- logger.info(f"Checking MySQL connection to {host}:{port}...")
- temp_connection = pymysql.connect(
- host=host,
- port=port,
- user=user,
- password=password,
- charset=charset,
- cursorclass=pymysql.cursors.DictCursor,
- connect_timeout=10 # 10 seconds timeout
- )
-
- # Test the connection by executing a simple query
- with temp_connection.cursor() as cursor:
- cursor.execute("SELECT VERSION() as version")
- result = cursor.fetchone()
- mysql_version = result.get('version', 'unknown') if result else 'unknown'
-
- logger.info(f"✓ MySQL connection successful (version: {mysql_version})")
- return True, ""
- except pymysql.err.OperationalError as e:
- # Safely get error code and message
- error_code = e.args[0] if len(e.args) > 0 else None
- error_msg = e.args[1] if len(e.args) > 1 else str(e)
- if error_code == 2003:
- error_message = f"Cannot connect to MySQL server at {host}:{port}. " \
- f"Please check if MySQL server is running and accessible."
- elif error_code == 1045:
- error_message = f"Access denied for user '{user}'@{host}. " \
- f"Please check MySQL username and password."
- else:
- error_message = f"MySQL connection error ({error_code}): {error_msg}"
- logger.error(f"✗ MySQL connection failed: {error_message}")
- return False, error_message
- except Exception as e:
- error_message = f"Unexpected error while checking MySQL connection: {str(e)}"
- logger.error(f"✗ MySQL connection check failed: {error_message}")
- return False, error_message
- finally:
- if temp_connection:
- temp_connection.close()
-
- @staticmethod
- def check_database_exists(db_name: str, connection=None) -> bool:
- """
- Check if a database exists
-
- Args:
- db_name: Database name to check
- connection: Optional existing MySQL connection
-
- Returns:
- True if database exists, False otherwise
- """
- temp_connection = None
- try:
- if connection is None:
- # Create temporary connection without specifying database
- temp_connection = pymysql.connect(
- host="localhost",
- port=3306,
- user="root",
- password="",
- charset="utf8mb4",
- cursorclass=pymysql.cursors.DictCursor
- )
- conn = temp_connection
- else:
- conn = connection
-
- with conn.cursor() as cursor:
- cursor.execute("SHOW DATABASES LIKE %s", (db_name,))
- result = cursor.fetchone()
- return result is not None
- except Exception as e:
- logger.error(f"Error checking database existence for {db_name}: {e}")
- return False
- finally:
- if temp_connection:
- temp_connection.close()
-
- @staticmethod
- def check_databases_exist(db_configs: List[DatabaseConfig]) -> Tuple[bool, List[str]]:
- """
- Check if all configured databases exist
- Each database config can use its own MySQL connection info
-
- Args:
- db_configs: List of DatabaseConfig objects
-
- Returns:
- Tuple of (all_exist: bool, missing_databases: List[str])
- """
- if not db_configs:
- return True, []
-
- missing_databases = []
-
- # Group database configs by connection info (host:port:user)
- # This allows us to check multiple databases on the same server efficiently
- connection_groups: Dict[Tuple[str, int, str], List[DatabaseConfig]] = {} # 使用DatabaseConfig类型
- for db_config in db_configs:
- # Use database-specific connection info only
- if not all([db_config.mysql_host, db_config.mysql_port, db_config.mysql_user]):
- missing_fields = []
- if not db_config.mysql_host:
- missing_fields.append("mysql_host")
- if not db_config.mysql_port:
- missing_fields.append("mysql_port")
- if not db_config.mysql_user:
- missing_fields.append("mysql_user")
- raise ValueError(f"Missing required database connection fields for {db_config.name}: {', '.join(missing_fields)}")
-
- host = db_config.mysql_host
- port = db_config.mysql_port
- user = db_config.mysql_user
- connection_key = (host, port, user)
-
- if connection_key not in connection_groups:
- connection_groups[connection_key] = []
- connection_groups[connection_key].append(db_config)
-
- # Check databases for each connection group
- for (host, port, user), configs in connection_groups.items():
- temp_connection = None
- try:
- # Get password and charset from first config in group (or use defaults)
- # Note: All configs in a group should use same password/charset for same host:port:user
- first_config = configs[0]
- password = first_config.mysql_password if first_config.mysql_password is not None else ""
- charset = first_config.mysql_charset if first_config.mysql_charset is not None else "utf8mb4"
-
- logger.info(f"Checking databases on MySQL server {host}:{port} (user: {user})...")
- temp_connection = pymysql.connect(
- host=host,
- port=port,
- user=user,
- password=password,
- charset=charset,
- cursorclass=pymysql.cursors.DictCursor,
- connect_timeout=10
- )
-
- # Get all existing databases on this server
- with temp_connection.cursor() as cursor:
- cursor.execute("SHOW DATABASES")
- existing_databases = {row['Database'] for row in cursor.fetchall()}
-
- # Check each configured database in this group
- for db_config in configs:
- if db_config.database not in existing_databases:
- missing_databases.append(
- f"{db_config.database} (name: {db_config.name}, server: {host}:{port})"
- )
- else:
- logger.info(f"✓ Database {db_config.database} exists on {host}:{port}")
-
- except Exception as e:
- # If we can't connect to check, mark all databases in this group as missing
- logger.error(f"Error checking databases on {host}:{port}: {e}")
- for db_config in configs:
- missing_databases.append(
- f"{db_config.database} (name: {db_config.name}, server: {host}:{port}, error: {str(e)})"
- )
- finally:
- if temp_connection:
- temp_connection.close()
-
- return len(missing_databases) == 0, missing_databases
-
- def _switch_database(self):
- """Switch to the target database using existing connection"""
- try:
- with self.connection.cursor() as cursor:
- cursor.execute(f"USE `{self.db_config.database}`")
- logger.info(f"Switched to MySQL database: {self.db_config.database} (name: {self.db_config.name})")
- except Exception as e:
- logger.error(f"Failed to switch to MySQL database {self.db_config.name}: {e}")
- raise
-
- def close(self):
- """Close MySQL connection"""
- if self.connection:
- self.connection.close()
- logger.info("MySQL connection closed")
-
- def _ensure_connection(self):
- """Ensure MySQL connection is alive and reconnect if necessary"""
- max_retries = 3
- for attempt in range(max_retries):
- try:
- # Check if connection is None
- if self.connection is None:
- logger.warning(f"MySQL connection for {self.db_config.name} is None, reconnecting... (attempt {attempt + 1}/{max_retries})")
- self._connect()
- # Verify connection was created
- if self.connection is None:
- raise RuntimeError(f"Failed to create connection for {self.db_config.name}")
- return
-
- # Check if connection is open
- if not hasattr(self.connection, 'open') or not self.connection.open:
- logger.warning(f"MySQL connection for {self.db_config.name} is closed, reconnecting... (attempt {attempt + 1}/{max_retries})")
- try:
- self.connection.close()
- except Exception:
- pass
- self.connection = None
- self._connect()
- # Verify connection was created
- if self.connection is None:
- raise RuntimeError(f"Failed to create connection for {self.db_config.name}")
- return
-
- # Test connection with ping
- # Note: ping(reconnect=True) may not work if connection is in bad state
- try:
- self.connection.ping(reconnect=False)
- except Exception:
- # Ping failed, connection is bad, recreate it
- logger.warning(f"Ping failed for {self.db_config.name}, recreating connection...")
- try:
- self.connection.close()
- except Exception:
- pass
- self.connection = None
- self._connect()
- if self.connection is None:
- raise RuntimeError(f"Failed to create connection for {self.db_config.name}")
- return
-
- return # Connection is good
- except (pymysql.err.OperationalError, pymysql.err.InterfaceError, AttributeError) as e:
- # Connection is dead or invalid, try to reconnect
- logger.warning(f"MySQL connection for {self.db_config.name} is dead/invalid, reconnecting... (attempt {attempt + 1}/{max_retries}): {e}")
- try:
- if self.connection:
- self.connection.close()
- except Exception:
- pass
- self.connection = None
- if attempt < max_retries - 1:
- import time
- time.sleep(0.5 * (attempt + 1)) # Exponential backoff
- else:
- # Last attempt, raise the error
- raise
- except Exception as e:
- logger.error(f"Unexpected error checking MySQL connection for {self.db_config.name}: {e}")
- # Set connection to None so next attempt will recreate it
- try:
- if self.connection:
- self.connection.close()
- except Exception:
- pass
- self.connection = None
- if attempt < max_retries - 1:
- import time
- time.sleep(0.5 * (attempt + 1))
- else:
- raise
-
- def fetch_all_documents(self) -> List[Dict]:
- """
- Fetch all documents from MySQL table
-
- Returns:
- List of document dictionaries with database source info
- """
- try:
- # Ensure connection is alive before using it
- self._ensure_connection()
-
- # Execute query with retry mechanism for connection issues
- max_retries = 3
- results = None
- for attempt in range(max_retries):
- try:
- # Ensure connection is still alive (may have been lost during previous attempt)
- if attempt > 0:
- self._ensure_connection()
- else:
- # Also ensure connection on first attempt (may have been set to None elsewhere)
- self._ensure_connection()
-
- # Double-check connection is not None before using
- if self.connection is None:
- logger.warning(f"Connection is None for {self.db_config.name} after _ensure_connection, retrying...")
- if attempt < max_retries - 1:
- import time
- time.sleep(0.5 * (attempt + 1))
- continue
- else:
- raise RuntimeError(f"Failed to establish connection for {self.db_config.name} after {max_retries} attempts")
-
- # Ensure we're using the correct database
- with self.connection.cursor() as cursor:
- cursor.execute(f"USE `{self.db_config.database}`")
-
- with self.connection.cursor() as cursor:
- # Get actual columns in the table
- cursor.execute(f"DESCRIBE `{self.db_config.table_name}`")
- table_columns = {row['Field'] for row in cursor.fetchall()}
-
- # Build query based on configured columns, only include columns that exist
- columns = []
- # Add ID column (required)
- if self.db_config.id_column in table_columns:
- columns.append(self.db_config.id_column)
- else:
- raise ValueError(f"ID column '{self.db_config.id_column}' does not exist in table {self.db_config.table_name}")
-
- # 添加所有 content 列（支持多个列），只添加存在的列
- for col in self.db_config.content_columns:
- if col in table_columns:
- columns.append(col)
- else:
- logger.warning(f"Content column '{col}' does not exist in table {self.db_config.table_name}, skipping")
-
- if not columns or all(col == self.db_config.id_column for col in columns):
- raise ValueError(f"No valid content columns found in table {self.db_config.table_name}")
-
- # Add title column if configured and exists
- if self.db_config.title_column and self.db_config.title_column in table_columns:
- columns.append(self.db_config.title_column)
- elif self.db_config.title_column:
- logger.warning(f"Title column '{self.db_config.title_column}' does not exist in table {self.db_config.table_name}, skipping")
-
- # Add metadata columns if specified, only add existing ones
- if self.db_config.metadata_columns:
- metadata_cols = [col.strip() for col in self.db_config.metadata_columns.split(",")]
- for col in metadata_cols:
- if col in table_columns:
- columns.append(col)
- else:
- logger.warning(f"Metadata column '{col}' does not exist in table {self.db_config.table_name}, skipping")
-
- query = f"SELECT {', '.join(columns)} FROM `{self.db_config.table_name}`"
- cursor.execute(query)
- results = cursor.fetchall()
- break # Success, exit retry loop
-
- except (pymysql.err.OperationalError, pymysql.err.InterfaceError) as e:
- # Safely get error code
- error_code = None
- error_msg = str(e)
- if hasattr(e, 'args') and e.args:
- try:
- error_code = e.args[0] if len(e.args) > 0 else None
- error_msg = e.args[1] if len(e.args) > 1 else str(e)
- except (IndexError, TypeError):
- error_code = None
-
- # Check for connection-related errors that should trigger reconnection
- should_reconnect = False
- if error_code == 2013: # Lost connection to MySQL server during query
- should_reconnect = True
- elif "Packet sequence number wrong" in error_msg or "Lost connection" in error_msg:
- should_reconnect = True
-
- if should_reconnect:
- logger.warning(f"Connection error during query for {self.db_config.name} ({error_msg}), reconnecting and retrying... (attempt {attempt + 1}/{max_retries})")
- # Close and recreate connection
- try:
- if self.connection:
- self.connection.close()
- except Exception:
- pass
- self.connection = None
-
- if attempt < max_retries - 1:
- import time
- time.sleep(0.5 * (attempt + 1)) # Exponential backoff
- continue
- else:
- logger.error(f"Failed to fetch documents after {max_retries} attempts: {e}")
- raise
- else:
- # Other operational error, raise immediately
- raise
- except Exception as e:
- # Non-connection errors, raise immediately
- raise
-
- if results is None:
- raise RuntimeError(f"Failed to fetch documents from {self.db_config.name} after {max_retries} attempts")
- logger.info(f"Fetched {len(results)} documents from MySQL database: {self.db_config.database} (name: {self.db_config.name})")
-
- # 如果配置了文件字段，则需根据文件字段的值来获取文件内容
- if self.db_config.file_column and self.db_config.file_column in self.db_config.content_columns:
- # 文件处理成功的文档
- processed_results = []
- # 失败文档ID列表
- failed_doc_ids = set()
- logger.info(
- f"MySQL database: {self.db_config.database} (table: {self.db_config.table_name}) has set file_column(value: {self.db_config.file_column}) field.")
- for doc in results:
- # 获取每个doc的文件字段值，即文件标识符
- file_column_value = doc.get(self.db_config.file_column)
- if not file_column_value or file_column_value is None:
- processed_results.append(doc)
- continue
- file_identifiers = [file_identifier.strip() for file_identifier in file_column_value.split(",")]
- try:
- file_content_parts = []
- for file_identifier in file_identifiers:
- # 调用接口获取文件字节流及文件名
- file_byte, file_name = self.fetch_file_bytes(file_identifier)
- if not file_byte:
- logger.warning(
- f"文档ID:{doc.get(self.db_config.id_column, '未知')}，文件标识符：{file_identifier} 获取字节流失败")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
- break
- # 调用文件读取方法获取文本内容
- file_text = self.file_parser.get_text_from_bytes(file_byte, file_name)
- if file_text and file_text.strip():
- file_content_parts.append(file_text.strip())
- logger.info(f"成功读取文件内容: {file_identifier} (长度: {len(file_text)})")
- else:
- logger.warning(f"文件标识符 {file_identifier} 提取的文本内容为空")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
- break
- if file_content_parts:
- file_content = "\n".join(file_content_parts)
- # 将文件内容添加到文档中
- doc[self.db_config.file_column] = file_content
- logger.info(
- f"文档 {doc.get(self.db_config.id_column, '未知')} 成功合并 {len(file_content_parts)} 个文件内容")
- # 将处理成功的文档添加到新列表中
- processed_results.append(doc)
- else:
- logger.warning(
- f"文档 {doc.get(self.db_config.id_column, '未知')} 没有成功提取任何文件内容")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
-
- except Exception as e:
- logger.error(
- f"处理文档({doc.get(self.db_config.id_column, '未知')}) 文件字段: {doc[self.db_config.file_column]} 时出错: {e}")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
- continue # 继续下一次doc循环
-
- # 记录失败文档统计信息
- if failed_doc_ids:
- failed_doc_ids_str = {str(doc_id) for doc_id in failed_doc_ids}
- ids_str = ', '.join(sorted(failed_doc_ids_str))
- logger.info(f"数据库配置[{self.db_config.name}]本次同步在处理文件过程中共有 {len(failed_doc_ids)} 个文档因文件处理失败被跳过: [{ids_str}]")
- else:
- logger.info(f"数据库配置[{self.db_config.name}]所有文档文件处理成功")
-
- results = processed_results
-
- # Add database source information to each document
- if results:
- for doc in results:
- doc['_db_source'] = self.db_config.name
- doc['_db_database'] = self.db_config.database
- doc['_db_table'] = self.db_config.table_name
-
- return results
- except Exception as e:
- logger.error(f"Error fetching documents from MySQL database {self.db_config.name}: {e}")
- raise
-
- def fetch_new_documents(self, last_sync_time: Optional[datetime] = None) -> List[Dict]:
- """
- Fetch new or updated documents since last sync
-
- Args:
- last_sync_time: Last synchronization timestamp
-
- Returns:
- List of new/updated document dictionaries with database source info
- """
- try:
- # Ensure connection is alive before using it
- self._ensure_connection()
-
- # Ensure we're using the correct database
- with self.connection.cursor() as cursor:
- cursor.execute(f"USE `{self.db_config.database}`")
-
- with self.connection.cursor() as cursor:
- # Get actual columns in the table
- cursor.execute(f"DESCRIBE `{self.db_config.table_name}`")
- table_columns = {row['Field'] for row in cursor.fetchall()}
-
- # Build query based on configured columns, only include columns that exist
- columns = []
- # Add ID column (required)
- if self.db_config.id_column in table_columns:
- columns.append(self.db_config.id_column)
- else:
- raise ValueError(f"ID column '{self.db_config.id_column}' does not exist in table {self.db_config.table_name}")
-
- # 添加所有 content 列（支持多个列），只添加存在的列
- for col in self.db_config.content_columns:
- if col in table_columns:
- columns.append(col)
- else:
- logger.warning(f"Content column '{col}' does not exist in table {self.db_config.table_name}, skipping")
-
- if not columns or all(col == self.db_config.id_column for col in columns):
- raise ValueError(f"No valid content columns found in table {self.db_config.table_name}")
-
- # Add title column if configured and exists
- if self.db_config.title_column and self.db_config.title_column in table_columns:
- columns.append(self.db_config.title_column)
- elif self.db_config.title_column:
- logger.warning(f"Title column '{self.db_config.title_column}' does not exist in table {self.db_config.table_name}, skipping")
-
- # Add metadata columns if specified, only add existing ones
- if self.db_config.metadata_columns:
- metadata_cols = [col.strip() for col in self.db_config.metadata_columns.split(",")]
- for col in metadata_cols:
- if col in table_columns:
- columns.append(col)
- else:
- logger.warning(f"Metadata column '{col}' does not exist in table {self.db_config.table_name}, skipping")
-
- query = f"SELECT {', '.join(columns)} FROM `{self.db_config.table_name}`"
-
- # Use updated_at_column if configured and last_sync_time is provided
- # Otherwise, fetch all documents (no incremental sync)
- if last_sync_time and self.db_config.updated_at_column:
- # Use configured updated_at column for incremental sync
- # MySQL DATETIME only has second precision, so we need to ensure
- # the comparison works correctly. Use >= to include records with same timestamp
- # and format the datetime to match MySQL's DATETIME format
- query += f" WHERE {self.db_config.updated_at_column} > %s"
- # Format datetime to MySQL DATETIME format (YYYY-MM-DD HH:MM:SS)
- # This ensures proper comparison with MySQL DATETIME type
- mysql_datetime = last_sync_time.strftime('%Y-%m-%d %H:%M:%S')
- cursor.execute(query, (mysql_datetime,))
- logger.info(
- f"Incremental sync using {self.db_config.updated_at_column}: "
- f"last_sync_time={last_sync_time} (formatted: {mysql_datetime}), "
- f"query: {query}"
- )
- else:
- # No updated_at column configured or no last_sync_time, fetch all
- if last_sync_time and not self.db_config.updated_at_column:
- logger.warning(
- f"No updated_at_column configured for {self.db_config.name}, "
- "fetching all documents for incremental sync"
- )
- cursor.execute(query)
-
- results = cursor.fetchall()
-
- # 如果配置了文件字段，则需根据文件字段的值来获取文件内容
- if self.db_config.file_column and self.db_config.file_column in self.db_config.content_columns:
- # 文件处理成功的文档
- processed_results = []
- # 失败文档ID列表
- failed_doc_ids = set()
- logger.info(
- f"MySQL database: {self.db_config.database} (table: {self.db_config.table_name}) has set file_column(value: {self.db_config.file_column}) field.")
- for doc in results:
- # 获取每个doc的文件字段值，即文件标识符
- file_column_value = doc.get(self.db_config.file_column)
- if not file_column_value or file_column_value is None:
- processed_results.append(doc)
- continue
- file_identifiers = [file_identifier.strip() for file_identifier in file_column_value.split(",")]
- try:
- file_content_parts = []
- for file_identifier in file_identifiers:
- # 调用接口获取文件字节流及文件名
- file_byte, file_name = self.fetch_file_bytes(file_identifier)
- if not file_byte:
- logger.warning(
- f"文档ID:{doc.get(self.db_config.id_column, '未知')}，文件标识符：{file_identifier} 获取字节流失败")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
- break
- # 调用文件读取方法获取文本内容
- file_text = self.file_parser.get_text_from_bytes(file_byte, file_name)
- if file_text and file_text.strip():
- file_content_parts.append(file_text.strip())
- logger.info(f"成功读取文件内容: {file_identifier} (长度: {len(file_text)})")
- else:
- logger.warning(f"文件标识符 {file_identifier} 提取的文本内容为空")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
- break
- if file_content_parts:
- file_content = "\n".join(file_content_parts)
- # 将文件内容添加到文档中
- doc[self.db_config.file_column] = file_content
- logger.info(
- f"文档 {doc.get(self.db_config.id_column, '未知')} 成功合并 {len(file_content_parts)} 个文件内容")
- # 将处理成功的文档添加到新列表中
- processed_results.append(doc)
- else:
- logger.warning(
- f"文档 {doc.get(self.db_config.id_column, '未知')} 没有成功提取任何文件内容")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
-
- except Exception as e:
- logger.error(
- f"处理文档({doc.get(self.db_config.id_column, '未知')}) 文件字段: {doc[self.db_config.file_column]} 时出错: {e}")
- failed_doc_ids.add(doc.get(self.db_config.id_column))
- continue # 继续下一次doc循环
-
- # 记录失败文档统计信息
- if failed_doc_ids:
- failed_doc_ids_str = {str(doc_id) for doc_id in failed_doc_ids}
- ids_str = ', '.join(sorted(failed_doc_ids_str))
- logger.warning(
- f"数据库配置[{self.db_config.name}]本次同步在处理文件过程中共有 {len(failed_doc_ids)} 个文档因文件处理失败被跳过: [{ids_str}]")
- else:
- logger.info(f"数据库配置[{self.db_config.name}]所有文档文件处理成功")
-
- results = processed_results
-
- # Add database source information to each document
- if results:
- for doc in results:
- doc['_db_source'] = self.db_config.name
- doc['_db_database'] = self.db_config.database
- doc['_db_table'] = self.db_config.table_name
-
- logger.info(f"Fetched {len(results)} new/updated documents from MySQL database: {self.db_config.database} (name: {self.db_config.name})")
- return results
- except Exception as e:
- logger.error(f"Error fetching new documents from MySQL database {self.db_config.name}: {e}")
- raise
-
- def get_document_by_id(self, doc_id: str) -> Optional[Dict]:
- """
- Get a single document by ID
-
- Args:
- doc_id: Document ID
-
- Returns:
- Document dictionary or None
- """
- try:
- # Ensure we're using the correct database
- with self.connection.cursor() as cursor:
- cursor.execute(f"USE `{self.db_config.database}`")
-
- with self.connection.cursor() as cursor:
- # Get actual columns in the table
- cursor.execute(f"DESCRIBE `{self.db_config.table_name}`")
- table_columns = {row['Field'] for row in cursor.fetchall()}
-
- # Build query based on configured columns, only include columns that exist
- columns = []
- # Add ID column (required)
- if self.db_config.id_column in table_columns:
- columns.append(self.db_config.id_column)
- else:
- raise ValueError(f"ID column '{self.db_config.id_column}' does not exist in table {self.db_config.table_name}")
-
- # 添加所有 content 列（支持多个列），只添加存在的列
- for col in self.db_config.content_columns:
- if col in table_columns:
- columns.append(col)
- else:
- logger.warning(f"Content column '{col}' does not exist in table {self.db_config.table_name}, skipping")
-
- if not columns or all(col == self.db_config.id_column for col in columns):
- raise ValueError(f"No valid content columns found in table {self.db_config.table_name}")
-
- # Add title column if configured and exists
- if self.db_config.title_column and self.db_config.title_column in table_columns:
- columns.append(self.db_config.title_column)
- elif self.db_config.title_column:
- logger.warning(f"Title column '{self.db_config.title_column}' does not exist in table {self.db_config.table_name}, skipping")
-
- # Add metadata columns if specified, only add existing ones
- if self.db_config.metadata_columns:
- metadata_cols = [col.strip() for col in self.db_config.metadata_columns.split(",")]
- for col in metadata_cols:
- if col in table_columns:
- columns.append(col)
- else:
- logger.warning(f"Metadata column '{col}' does not exist in table {self.db_config.table_name}, skipping")
-
- query = f"SELECT {', '.join(columns)} FROM `{self.db_config.table_name}` WHERE {self.db_config.id_column} = %s"
- cursor.execute(query, (doc_id,))
- result = cursor.fetchone()
- if result:
- result['_db_source'] = self.db_config.name
- result['_db_database'] = self.db_config.database
- result['_db_table'] = self.db_config.table_name
- return result
- except Exception as e:
- logger.error(f"Error fetching document {doc_id} from MySQL database {self.db_config.name}: {e}")
- return None
-
- # 获取文件字节流及文件名
- def fetch_file_bytes(self, file_identifier: str) -> tuple[bytes, str]:
- """
- 获取文件流、文件名
- Args:
- file_identifier: 文件标识值
- Returns:
- tuple: (文件字节流, 文件名)
- """
- # 获取文件源类型配置
- file_source_type = self.db_config.file_source_type
-
- if file_source_type == "filesystem":
- # 从本地文件系统读取文件
- import os
- try:
- if not self.db_config.file_system_base_path:
- raise ValueError("file_system_base_path is required for filesystem source type")
-
- file_path = os.path.join(self.db_config.file_system_base_path, file_identifier)
- if not os.path.exists(file_path):
- raise FileNotFoundError(f"File not found: {file_path}")
-
- with open(file_path, 'rb') as f:
- file_bytes = f.read()
-
- file_name = os.path.basename(file_path)
- logger.info(f"Successfully fetched file from local filesystem: {file_path} (size: {len(file_bytes)} bytes)")
- return file_bytes, file_name
- except Exception as e:
- logger.error(f"Error fetching file from local filesystem: {e}")
- raise
-
- elif file_source_type == "scp":
- # 从远程文件系统通过SCP读取文件
- import os
- try:
- if not self.db_config.file_system_base_path:
- raise ValueError("file_system_base_path is required for scp source type")
- if not self.db_config.scp_host:
- raise ValueError("scp_host is required for scp source type")
-
- # 导入paramiko库
- import paramiko
-
- # 创建SSH客户端
- ssh_client = paramiko.SSHClient()
- ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
- # 连接到远程主机
- connect_kwargs = {
- 'hostname': self.db_config.scp_host,
- 'port': self.db_config.scp_port,
- 'username': self.db_config.scp_username
- }
-
- if self.db_config.scp_key_path:
- connect_kwargs['key_filename'] = self.db_config.scp_key_path
- elif self.db_config.scp_password:
- connect_kwargs['password'] = self.db_config.scp_password
- else:
- raise ValueError("Either scp_key_path or scp_password must be provided for SCP")
-
- ssh_client.connect(**connect_kwargs)
-
- # 创建SCP客户端
- scp_client = ssh_client.open_sftp()
-
- # 构建远程文件路径
- remote_file_path = os.path.join(self.db_config.file_system_base_path, file_identifier)
-
- # 读取文件内容
- with scp_client.open(remote_file_path, 'rb') as f:
- file_bytes = f.read()
-
- file_name = os.path.basename(remote_file_path)
- logger.info(f"Successfully fetched file from remote filesystem via SCP: {remote_file_path} (size: {len(file_bytes)} bytes)")
-
- # 关闭连接
- scp_client.close()
- ssh_client.close()
-
- return file_bytes, file_name
- except Exception as e:
- logger.error(f"Error fetching file from remote filesystem via SCP: {e}")
- raise
-
- else: # api
- # 从API获取文件
- base_url = settings.FILE_DOWNLOAD_BASE_URL
- url = f"{base_url}/{file_identifier}"
- try:
- # 发送GET请求获取文件流
- response = requests.get(url, stream=True, timeout=30)
-
- # 检查响应内容是否为文件
- content_type = response.headers.get('Content-Type', '')
- if 'application/json' in content_type:
- # 如果是JSON响应，尝试解析错误信息
- error_data = response.json()
- error_msg = error_data.get('msg', '未知错误')
- error_code = error_data.get('code', '未知')
- raise ValueError(f"API返回错误: {error_msg} (代码: {error_code})")
- response.raise_for_status() # 如果请求失败，抛出HTTPError异常
- # 获取文件字节流
- file_bytes = response.content
- logger.info(f"Successfully fetch file bytes from API, file: {file_identifier} (size: {len(file_bytes)} bytes)")
-
- # 从响应头中获取文件名
- file_name = "unknown"
-
- # 从 Content-Disposition 头获取文件名
- content_disposition = response.headers.get('Content-Disposition', '')
- if 'filename=' in content_disposition:
- # 提取文件名，处理可能的引号和多个filename参数
- import re
- # match = re.search(r'filename=["\']?([^"\']+)["\']?', content_disposition)
- match = re.search(r'filename=["\']?([^;"\']+)["\']?', content_disposition)
- if match:
- file_name = match.group(1)
- # 处理URL编码的文件名
- from urllib.parse import unquote
- file_name = unquote(file_name)
- logger.info(f"Successfully get filename: {file_name}")
- return file_bytes, file_name
- except requests.exceptions.RequestException as e:
- logger.error(f"Error fetching file from API: {e}")
- raise
- except Exception as e:
- logger.error(f"Unexpected error fetching file from API: {e}")
- raise
\ No newline at end of file
diff --git a/docker-compose.yml b/docker-compose.yml
index 5b053ec..a63fa85 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -57,8 +57,6 @@ services:
env_file:
- .env
volumes:
- # 挂载配置文件目录（支持动态增加配置）
- - ./configs:/app/configs:rw
# 挂载日志目录（可选）
- ./logs:/app/logs
# 挂载本地 data 目录以持久化会话/SQLite 数据库
@@ -87,7 +85,6 @@ services:
- MYSQL_USER=${MYSQL_USER:-root}
- MYSQL_PASSWORD=${MYSQL_PASSWORD:-}
- MYSQL_DATABASE=${MYSQL_DATABASE:-forgeplus}
- - MYSQL_CHARSET=${MYSQL_CHARSET:-utf8mb4}
# 多数据库配置（可选）
- MYSQL_DATABASES_CONFIG=${MYSQL_DATABASES_CONFIG:-}
diff --git a/rag/file_parser.py b/rag/file_parser.py
index 57bfb04..7ff858b 100644
--- a/rag/file_parser.py
+++ b/rag/file_parser.py
@@ -31,10 +31,10 @@ class FileParser:
def is_supported(self, filename: str) -> bool:
"""
- Check if file format is supported
+ Check if the file format is supported by the parser
Args:
- filename: File name or path
+ filename: Name or path of the file to check
Returns:
True if supported, False otherwise
@@ -42,14 +42,44 @@ class FileParser:
ext = Path(filename).suffix.lower() # 获取文件类型并转为小写
return ext in self.SUPPORTED_EXTENSIONS
- def parse_file(self, file_path: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None) -> List[Document]:
+ def _generate_doc_id(self, filename: str, host: Optional[str] = None) -> str:
+ """
+ Generate a unique doc_id based on host address and filename
+
+ Args:
+ filename: Name of the file
+ host: Optional host address (if not provided, will use local machine IP)
+
+ Returns:
+ Unique doc_id generated from host address and filename
+ """
+ # 使用传入的host参数（如果提供），否则获取本地机器IP地址
+ if host:
+ host_address = host
+ else:
+ import socket
+ try:
+ # 获取机器IP地址
+ hostname = socket.gethostname()
+ host_address = socket.gethostbyname(hostname)
+ except:
+ # 如果无法获取IP，使用默认值
+ host_address = 'unknown'
+
+ # 使用主机地址和完整文件路径生成doc_id，确保唯一性
+ # 替换路径中的特殊字符，避免生成无效的doc_id
+ sanitized_path = filename.replace('/', '_').replace('\\', '_').replace(':', '_').replace(' ', '_')
+ return f"{host_address}_{sanitized_path}"
+
+ def parse_file(self, file_path: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None, host: Optional[str] = None) -> List[Document]:
"""
Parse a file into LlamaIndex Documents
Args:
file_path: Path to the file
- doc_id: Optional document ID (if not provided, will use filename)
+ doc_id: Optional document ID (if not provided, will use filename and host)
metadata: Optional metadata to add to documents
+ host: Optional host address (if not provided, will use local machine IP)
Returns:
List of LlamaIndex Document objects
@@ -63,16 +93,47 @@ class FileParser:
if not self.is_supported(file_path):
raise ValueError(f"Unsupported file format: {ext}. Supported formats: {', '.join(self.SUPPORTED_EXTENSIONS)}")
- # Use doc_id or generate from filename
+ # 生成基于主机地址和真实文件名的doc_id（如果未提供）
if not doc_id:
- doc_id = Path(filename).stem
+ doc_id = self._generate_doc_id(filename, host)
try:
- logger.info(f"Starting to parse file: {filename} (type: {ext})")
+ logger.info(f"Starting to parse file: {filename} (type: {ext})\n")
+
+ # 处理 .doc 文件：通过 soffice-service 转换为 .docx
+ if ext == '.doc':
+ logger.info(f"检测到 .doc 文件，开始转换为 .docx 格式: {file_path}")
+ # 读取 .doc 文件内容
+ with open(file_path, 'rb') as f:
+ file_bytes = f.read()
+ # 上传文件到 soffice-service 的 convert 接口
+ import requests
+ from config import settings
+ files = {'file': (filename, file_bytes, 'application/msword')}
+ soffice_url = f"http://{settings.SOFFICE_HOST}:{settings.SOFFICE_PORT}/convert"
+ response = requests.post(soffice_url, files=files, timeout=60)
+ try:
+ response.raise_for_status() # 检查 HTTP 错误
+ except requests.exceptions.HTTPError as e:
+ logger.error(f"转换请求失败: {e}")
+ raise
+ # 将转换后的 docx 内容保存到临时文件进行解析
+ import tempfile
+ with tempfile.NamedTemporaryFile(delete=False, suffix='.docx') as tmp_docx_file:
+ tmp_docx_file.write(response.content)
+ tmp_path = tmp_docx_file.name
+ # 使用转换后的临时文件路径
+ parse_path = tmp_path
+ cleanup_tmp = True
+ else:
+ # 对于其他文件格式，直接使用原始文件路径
+ parse_path = file_path
+ cleanup_tmp = False
+
# Use LlamaIndex's SimpleDirectoryReader for parsing
# It supports many formats out of the box
reader = SimpleDirectoryReader(
- input_files=[file_path],
+ input_files=[parse_path],
filename_as_id=False # We'll set custom IDs
)
@@ -131,14 +192,23 @@ class FileParser:
if not valid_documents:
raise ValueError(f"No valid documents extracted from {filename}. All documents are empty.")
- logger.info(f"Successfully parsed file {filename}: {len(valid_documents)} valid document(s) (total: {len(documents)}, skipped empty: {len(documents) - len(valid_documents)})")
+ logger.info(f"Successfully parsed file {filename}: {len(valid_documents)} valid document(s) (total: {len(documents)}, skipped empty: {len(documents) - len(valid_documents)})\n")
return valid_documents
except Exception as e:
logger.error(f"Error parsing file {file_path}: {e}", exc_info=True)
raise
+ finally:
+ # 清理临时文件（如果有）
+ if 'cleanup_tmp' in locals() and cleanup_tmp:
+ try:
+ import os
+ os.unlink(tmp_path)
+ logger.debug(f"Cleaned up temporary file: {tmp_path}")
+ except Exception as e:
+ logger.warning(f"Failed to delete temporary file {tmp_path}: {e}")
- def parse_file_content(self, content: bytes, filename: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None) -> List[Document]:
+ def parse_file_content(self, content: bytes, filename: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None, host: Optional[str] = None) -> List[Document]:
"""
Parse file content from bytes into LlamaIndex Documents
@@ -147,6 +217,7 @@ class FileParser:
filename: Original filename (for format detection)
doc_id: Optional document ID
metadata: Optional metadata to add to documents
+ host: Optional host address (for remote files, use the remote host address)
Returns:
List of LlamaIndex Document objects
@@ -163,6 +234,10 @@ class FileParser:
else:
ext = '.txt'
+ # 生成基于主机地址和真实文件名的doc_id（如果未提供）
+ if not doc_id:
+ doc_id = self._generate_doc_id(filename, host)
+
# Use tempfile to save content and parse
with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp_file:
tmp_file.write(content)
@@ -180,75 +255,4 @@ class FileParser:
logger.warning(f"Failed to delete temporary file {tmp_path}: {e}")
- def get_text_from_bytes(self, file_bytes: bytes, file_name: str) -> str:
- """
- 根据文件的字节流获取文件内的文本字符串
- Args:
- file_bytes: 文件字节流
- file_name: 文件名
- Returns:
- 文件内容的文本字符串
- """
- import tempfile
- # 获取文件类型
- if file_name == "unknown":
- ext = '.txt'
- else:
- ext = Path(file_name).suffix.lower()
-
- # 初始化临时文件路径
- tmp_path = None
-
- try:
- # 检查是否为.doc文件，如果是则调用soffice-service转换为.docx
- if ext == '.doc':
- logger.info(f"检测到.doc文件，开始转换为.docx格式: {file_name}")
- # 上传文件到 soffice-service的 convert接口
- files = {'file': (file_name, file_bytes, 'application/msword')}
- soffice_url = f"http://{settings.SOFFICE_HOST}:{settings.SOFFICE_PORT}/convert"
- response = requests.post(soffice_url, files=files, timeout=60)
- try:
- response.raise_for_status() # 检查 HTTP错误
- except requests.exceptions.HTTPError as e:
- logger.error(f"转换请求失败: {e}")
- raise
- # 将转换后的 docx 内容保存到临时文件进行解析
- with tempfile.NamedTemporaryFile(delete=False, suffix='.docx') as tmp_docx_file:
- tmp_docx_file.write(response.content)
- tmp_path = tmp_docx_file.name
- else:
- # Use tempfile to save content and parse
- with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp_file:
- tmp_file.write(file_bytes)
- tmp_path = tmp_file.name
-
- # Parse the temporary file
- documents = self.parse_file(tmp_path, doc_id=None)
- # 提取所有非空文档的文本内容
- text_parts = []
- for i, doc in enumerate(documents):
- if doc.text and doc.text.strip():
- text_parts.append(doc.text.strip())
- logger.debug(f"提取到文档 {i} 的文本内容，长度: {len(doc.text)}")
- else:
- logger.warning(f"文档 {i} 内容为空，已跳过")
-
- if not text_parts:
- raise ValueError("所有文档内容均为空，无法提取有效文本")
-
- # 合并所有文本内容
- full_text = "\n\n".join(text_parts)
- logger.info(f"成功从文件(文件名：{file_name})字节流中提取文本，总长度: {len(full_text)} 字符")
- return full_text
-
- except Exception as e:
- logger.error(f"解析文件(文件名：{file_name})字节流时出错: {e}", exc_info=True)
- raise
- finally:
- # Clean up temporary file
- if tmp_path is not None:
- try:
- os.unlink(tmp_path)
- except Exception as e:
- logger.warning(f"Failed to delete temporary file {tmp_path}: {e}")
diff --git a/scan_databases.py b/scan_databases.py
deleted file mode 100644
index 4c30795..0000000
--- a/scan_databases.py
+++ /dev/null
@@ -1,402 +0,0 @@
-#!/usr/bin/env python3
-"""
-扫描 MySQL 数据库，分析表结构并生成配置
-自动识别包含内容列的表（include 模式）：
-- 只考虑表中包含 text、longtext、mediumtext、tinytext 类型的列
-- 或者表中包含 blob、longblob、mediumblob、tinyblob 类型的列
-支持通过 scan_exclude_config.json 配置文件额外排除不需要的数据库和表
-"""
-import pymysql
-import json
-import os
-import re
-from typing import List, Dict, Any, Set
-from config import settings
-
-def load_exclude_config(config_path: str = 'scan_exclude_config.json') -> Dict[str, Any]:
- """
- 加载排除配置文件
-
- Args:
- config_path: 配置文件路径
-
- Returns:
- 排除配置字典，包含 exclude_databases, exclude_databases_patterns,
- 和 exclude_tables_patterns（仅支持正则表达式模式）
- """
- default_config = {
- "exclude_databases": ["information_schema", "performance_schema", "mysql", "sys"],
- "exclude_databases_patterns": [],
- "exclude_tables_patterns": {}
- }
-
- if not os.path.exists(config_path):
- print(f"排除配置文件不存在: {config_path}，使用默认配置")
- return default_config
-
- try:
- with open(config_path, 'r', encoding='utf-8') as f:
- config = json.load(f)
- # 确保配置格式正确
- if 'exclude_databases' not in config:
- config['exclude_databases'] = default_config['exclude_databases']
- if 'exclude_databases_patterns' not in config:
- config['exclude_databases_patterns'] = []
- if 'exclude_tables_patterns' not in config:
- config['exclude_tables_patterns'] = {}
- return config
- except Exception as e:
- print(f"加载排除配置文件失败: {e}，使用默认配置")
- return default_config
-
-def matches_pattern(name: str, patterns: List[str]) -> bool:
- """
- 检查名称是否匹配任何正则表达式模式
-
- Args:
- name: 要检查的名称
- patterns: 正则表达式模式列表
-
- Returns:
- 如果匹配任何模式返回 True，否则返回 False
- """
- for pattern in patterns:
- try:
- if re.search(pattern, name):
- return True
- except re.error as e:
- print(f"警告: 正则表达式模式 '{pattern}' 无效: {e}")
- continue
- return False
-
-def get_mysql_connection(host="localhost", port=3306, user="root", password="", charset="utf8mb4"):
- """获取 MySQL 连接"""
- return pymysql.connect(
- host=host,
- port=port,
- user=user,
- password=password,
- charset=charset,
- cursorclass=pymysql.cursors.DictCursor
- )
-
-def get_databases(conn, exclude_databases: Set[str] = None, exclude_patterns: List[str] = None) -> List[str]:
- """
- 获取所有数据库列表
-
- Args:
- conn: MySQL 连接
- exclude_databases: 要排除的数据库集合（精确匹配）
- exclude_patterns: 要排除的数据库正则表达式模式列表
-
- Returns:
- 数据库列表
- """
- if exclude_databases is None:
- exclude_databases = set()
- if exclude_patterns is None:
- exclude_patterns = []
-
- with conn.cursor() as cursor:
- cursor.execute("SHOW DATABASES")
- databases = [row['Database'] for row in cursor.fetchall()]
-
- # 过滤掉排除的数据库
- filtered_databases = []
- for db in databases:
- # 检查精确匹配
- if db in exclude_databases:
- continue
- # 检查正则表达式匹配
- if matches_pattern(db, exclude_patterns):
- continue
- filtered_databases.append(db)
-
- return filtered_databases
-
-def get_tables(conn, database: str) -> List[str]:
- """获取指定数据库的所有表"""
- with conn.cursor() as cursor:
- cursor.execute(f"USE `{database}`")
- cursor.execute("SHOW TABLES")
- # 表名在结果中的键名是 'Tables_in_{database}'
- key = f"Tables_in_{database}"
- return [row[key] for row in cursor.fetchall()]
-
-def get_table_columns(conn, database: str, table: str) -> List[Dict[str, Any]]:
- """获取表的列信息"""
- with conn.cursor() as cursor:
- cursor.execute(f"USE `{database}`")
- cursor.execute(f"DESCRIBE `{table}`")
- return cursor.fetchall()
-
-def should_include_table(table_name: str, columns: List[Dict[str, Any]]) -> bool:
- """
- 判断表是否应该被包含（仅根据列类型判断，并排除敏感信息表）
- 只考虑 text、longtext、mediumtext、tinytext 以及 blob 等类型的字段
- 排除包含敏感信息的表（authentication、private、key、secret、password 等）
-
- Args:
- table_name: 表名
- columns: 表的列信息列表
-
- Returns:
- 如果表应该被包含返回 True，否则返回 False
- """
- # 排除敏感信息表（检查表名）
- table_lower = table_name.lower()
- sensitive_keywords = ['authentication', 'auth', 'private', 'key', 'secret', 'password',
- 'pwd', 'token', 'credential', 'credential', 'session', 'login',
- 'user_password', 'user_secret', 'api_key', 'access_key', 'secret_key']
-
- # 检查表名是否包含敏感关键词
- if any(keyword in table_lower for keyword in sensitive_keywords):
- return False
-
- # 检查列名是否包含敏感关键词
- column_names = [col['Field'].lower() for col in columns]
- if any(keyword in col_name for col_name in column_names for keyword in sensitive_keywords):
- return False
-
- column_types = [col['Type'].lower() for col in columns]
-
- # 只检查是否有文本类型列（text、longtext、mediumtext、tinytext、blob、longblob、mediumblob、tinyblob）
- text_types = ['text', 'longtext', 'mediumtext', 'tinytext', 'blob', 'longblob', 'mediumblob', 'tinyblob']
- has_text_type = any(any(text_type in col_type for text_type in text_types) for col_type in column_types)
-
- return has_text_type
-
-def analyze_table_structure(columns: List[Dict[str, Any]]) -> Dict[str, Any]:
- """分析表结构，推断配置项"""
- column_names = [col['Field'] for col in columns]
- column_types = {col['Field']: col['Type'].lower() for col in columns}
-
- # 查找 ID 列（通常是 id, *_id）
- id_column = None
- for col in column_names:
- if col.lower() == 'id':
- id_column = col
- break
- elif col.lower().endswith('_id'):
- id_column = col
- break
-
- # 查找内容列（只考虑 text、longtext、mediumtext、tinytext 以及 blob 等类型）
- text_types = ['text', 'longtext', 'mediumtext', 'tinytext', 'blob', 'longblob', 'mediumblob', 'tinyblob']
- content_columns = []
- for col in column_names:
- col_type = column_types.get(col, '').lower()
- # 只包含 text 和 blob 类型的列
- if any(text_type in col_type for text_type in text_types):
- if col != id_column and not col.lower().endswith('_id'):
- content_columns.append(col)
-
- # 查找标题列（可能是 title, subject, name）
- title_column = None
- title_keywords = ['title', 'subject', 'name']
- for col in column_names:
- if col.lower() in title_keywords:
- title_column = col
- break
-
- # 查找更新时间列（可能是 updated_at, updated_on, modified_at, update_time）
- updated_at_column = None
- update_keywords = ['updated_at', 'updated_on', 'modified_at', 'update_time', 'updated_time']
- for col in column_names:
- if col.lower() in update_keywords:
- updated_at_column = col
- break
-
- # 查找元数据列（author, category 等，不包括时间相关列）
- metadata_columns = []
- # 时间相关的关键词，这些不应该出现在 metadata_columns 中
- time_keywords = ['created_at', 'created_on', 'created_time', 'create_time', 'create_by',
- 'updated_at', 'updated_on', 'updated_time', 'update_time', 'modified_at',
- 'created_unix', 'updated_unix']
- # 元数据关键词（不包括时间相关）
- metadata_keywords = ['author', 'category', 'tag', 'tags', 'status', 'type', 'priority', 'source']
-
- for col in column_names:
- col_lower = col.lower()
- # 排除时间相关的列
- if any(time_keyword in col_lower for time_keyword in time_keywords):
- continue
- # 排除以 created_ 或 create_ 开头的列（时间相关）
- if col_lower.startswith('created_') or col_lower.startswith('create_'):
- continue
- # 排除以 updated_ 或 update_ 开头的列（时间相关）
- if col_lower.startswith('updated_') or col_lower.startswith('update_'):
- continue
- # 排除以 modified_ 开头的列（时间相关）
- if col_lower.startswith('modified_'):
- continue
- # 添加符合条件的元数据列
- if col_lower in metadata_keywords:
- metadata_columns.append(col)
-
- return {
- 'id_column': id_column or 'id',
- 'content_columns': content_columns[:3] if content_columns else ['content'], # 最多取3个
- 'title_column': title_column,
- 'updated_at_column': updated_at_column,
- 'metadata_columns': metadata_columns[:5] if metadata_columns else None # 最多取5个
- }
-
-def generate_config(conn, exclude_config: Dict[str, Any] = None) -> List[Dict[str, Any]]:
- """
- 生成配置列表
-
- Args:
- conn: MySQL 连接
- exclude_config: 排除配置字典
-
- Returns:
- 配置列表
- """
- if exclude_config is None:
- exclude_config = {
- "exclude_databases": [],
- "exclude_databases_patterns": [],
- "exclude_tables_patterns": {}
- }
-
- configs = []
- exclude_databases = set(exclude_config.get('exclude_databases', []))
- exclude_db_patterns = exclude_config.get('exclude_databases_patterns', [])
- exclude_table_patterns = exclude_config.get('exclude_tables_patterns', {})
-
- databases = get_databases(conn, exclude_databases, exclude_db_patterns)
-
- print(f"找到 {len(databases)} 个数据库: {', '.join(databases)}")
- if exclude_databases:
- print(f"排除的数据库（精确匹配）: {', '.join(sorted(exclude_databases))}")
- if exclude_db_patterns:
- print(f"排除的数据库（正则表达式）: {', '.join(exclude_db_patterns)}")
-
- for database in databases:
- print(f"\n扫描数据库: {database}")
- try:
- tables = get_tables(conn, database)
- # 获取该数据库要排除的表（正则表达式模式）
- db_exclude_patterns = exclude_table_patterns.get(database, [])
-
- if db_exclude_patterns:
- print(f" 排除的表（正则表达式）: {', '.join(db_exclude_patterns)}")
-
- # 过滤掉排除的表（仅使用正则表达式模式）
- filtered_tables = []
- for table in tables:
- # 检查正则表达式匹配
- if matches_pattern(table, db_exclude_patterns):
- continue
- filtered_tables.append(table)
-
- print(f" 找到 {len(tables)} 个表，排除后 {len(filtered_tables)} 个表")
-
- # 进一步过滤：只包含有内容列的表
- included_tables = []
- excluded_by_content = []
-
- for table in filtered_tables:
- try:
- columns = get_table_columns(conn, database, table)
- if should_include_table(table, columns):
- included_tables.append(table)
- else:
- excluded_by_content.append(table)
- except Exception as e:
- print(f" 检查表 {table} 时出错: {e}")
- continue
-
- print(f" 自动识别包含内容列的表: {len(included_tables)} 个")
- if excluded_by_content:
- print(f" 排除无内容列的表: {len(excluded_by_content)} 个")
- if len(excluded_by_content) <= 10:
- print(f" {', '.join(excluded_by_content)}")
- else:
- print(f" {', '.join(excluded_by_content[:10])} ... (共 {len(excluded_by_content)} 个)")
-
- for table in included_tables:
- print(f" 分析表: {table}")
- try:
- columns = get_table_columns(conn, database, table)
- analysis = analyze_table_structure(columns)
-
- # 生成配置项
- config = {
- 'name': f"{database}_{table}",
- 'database': database,
- 'table_name': table,
- 'id_column': analysis['id_column'],
- 'content_column': ','.join(analysis['content_columns']),
- 'title_column': analysis['title_column'],
- 'metadata_columns': ','.join(analysis['metadata_columns']) if analysis['metadata_columns'] else None,
- 'content_separator': '\n',
- 'updated_at_column': analysis['updated_at_column']
- }
-
- # 清理 None 值
- config = {k: v for k, v in config.items() if v is not None}
- configs.append(config)
-
- print(f" ID列: {analysis['id_column']}")
- print(f" 内容列: {', '.join(analysis['content_columns'])}")
- if analysis['title_column']:
- print(f" 标题列: {analysis['title_column']}")
- if analysis['updated_at_column']:
- print(f" 更新时间列: {analysis['updated_at_column']}")
- if analysis['metadata_columns']:
- print(f" 元数据列: {', '.join(analysis['metadata_columns'])}")
-
- except Exception as e:
- print(f" 错误: {e}")
- continue
-
- except Exception as e:
- print(f" 错误: {e}")
- continue
-
- return configs
-
-def main():
- """主函数"""
- try:
- # 加载排除配置
- exclude_config = load_exclude_config()
-
- conn = get_mysql_connection()
- print("成功连接到 MySQL 服务器")
- print("=" * 60)
- print("扫描模式: 自动识别包含内容列的表（include 模式）")
- print("识别条件:")
- print(" - 表中包含 text、longtext、mediumtext、tinytext 类型的列")
- print(" - 或者表中包含 blob、longblob、mediumblob、tinyblob 类型的列")
- print(" - 自动排除包含敏感信息的表（authentication、private、key、secret、password 等）")
- print("=" * 60)
- if os.path.exists('scan_exclude_config.json'):
- print(f"额外排除配置: scan_exclude_config.json")
- print()
-
- configs = generate_config(conn, exclude_config)
-
- conn.close()
-
- # 输出 JSON 配置
- print(f"\n\n生成 {len(configs)} 个配置项")
- print("\n配置 JSON:")
- print(json.dumps(configs, indent=2, ensure_ascii=False))
-
- # 保存到文件
- output_file = 'databases_config_new.json'
- with open(output_file, 'w', encoding='utf-8') as f:
- json.dump(configs, f, indent=2, ensure_ascii=False)
- print(f"\n配置已保存到: {output_file}")
-
- except Exception as e:
- print(f"错误: {e}")
- import traceback
- traceback.print_exc()
-
-if __name__ == '__main__':
- main()
-
diff --git a/static/config/index.html b/static/config/index.html
index 35e1400..803848a 100644
--- a/static/config/index.html
+++ b/static/config/index.html
@@ -4,8 +4,8 @@
RAG 配置管理
-
-
+
+
@@ -13,7 +13,15 @@