diff --git a/api/main.py b/api/main.py
index 6a7fd38..ec233a5 100644
--- a/api/main.py
+++ b/api/main.py
@@ -2,9 +2,6 @@
FastAPI main application
"""
import warnings
-# Suppress pkg_resources deprecation warning from debugpy extension
-# This warning is harmless and comes from VS Code/Cursor debugpy extension
-# It appears when debugpy is loaded, which happens during debugging
warnings.filterwarnings("ignore", message=".*pkg_resources is deprecated.*", category=UserWarning)
from contextlib import asynccontextmanager
@@ -17,7 +14,8 @@ from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from loguru import logger
from config import settings
-from rag import VectorStoreManager, RAGEngine, FileParser, DocumentProcessor
+from db_utils import get_db_connection, init_session_db
+from rag import VectorStoreManager, RAGEngine, FileParser
from sync_service import SyncServiceManager
import requests
from datetime import datetime
@@ -35,6 +33,7 @@ import re
import markdown2
from pathlib import Path
import hashlib
+from sync.base_sync import BaseSync
# Global instances
@@ -42,14 +41,13 @@ vector_store_manager: Optional[VectorStoreManager] = None
rag_engine: Optional[RAGEngine] = 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
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup and shutdown events"""
- global vector_store_manager, rag_engine, sync_manager, file_parser, document_processor, auto_sync_task
+ global vector_store_manager, rag_engine, sync_manager, file_parser, auto_sync_task
# Startup
try:
@@ -79,13 +77,6 @@ async def lifespan(app: FastAPI):
logger.error(f"Failed to initialize FileParser: {e}")
raise
- # Initialize document processor
- try:
- document_processor = DocumentProcessor()
- logger.info("✓ DocumentProcessor initialized")
- except Exception as e:
- logger.error(f"Failed to initialize DocumentProcessor: {e}")
- raise
logger.info("✓ Core RAG services initialized")
@@ -368,36 +359,7 @@ def hash_password(password: str) -> str:
return hashlib.sha256(password.encode()).hexdigest()
-def init_session_db():
- conn = sqlite3.connect(DB_PATH)
- try:
- conn.execute(
- """
- CREATE TABLE IF NOT EXISTS users (
- id TEXT PRIMARY KEY,
- username TEXT UNIQUE,
- password TEXT,
- create_time TEXT
- )
- """
- )
- conn.execute(
- """
- CREATE TABLE IF NOT EXISTS sessions (
- id TEXT PRIMARY KEY,
- user_login TEXT,
- title TEXT,
- data TEXT,
- update_time TEXT
- )
- """
- )
- # Improve concurrency for writes
- conn.execute("PRAGMA journal_mode=WAL;")
- conn.execute("PRAGMA synchronous=NORMAL;")
- conn.commit()
- finally:
- conn.close()
+
def load_sessions():
@@ -1049,41 +1011,23 @@ async def manual_sync(request: SyncRequest):
try:
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}"
+ 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=True, is_manual=True)
+ message = f"Full sync completed for {request.source_name}" + (" (forced re-processing)" if request.force else "")
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"
+ await specific_sync_service.sync_incremental(is_manual=True)
+ message = f"Incremental sync completed for {request.source_name}"
+
else:
# 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"
-
+ # 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"
+
return {"status": "success", "message": message}
except Exception as e:
logger.error(f"Error during sync: {e}")
@@ -1153,7 +1097,7 @@ async def upload_document(
HTTPException: 413 if file size exceeds 10MB
HTTPException: 400 if document parsing fails
"""
- if file_parser is None or vector_store_manager is None or document_processor is None:
+ if file_parser is None or vector_store_manager is None:
raise HTTPException(status_code=503, detail="Services not initialized")
try:
@@ -1217,7 +1161,7 @@ async def upload_document(
# Chunk documents if needed
logger.info(f"Chunking {len(documents)} document(s)...")
- chunked_documents = document_processor.chunk_documents(documents)
+ chunked_documents = BaseSync.chunk_documents(documents)
logger.info(f"Chunked into {len(chunked_documents)} chunk(s)")
# Add to vector store (this may take time for large documents due to embedding generation)
@@ -1401,20 +1345,10 @@ async def create_config(config: Dict[str, Any]):
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":
- # 远程文件夹配置需要：主机、文件夹路径
+ elif config_type == "folder":
+ # 文件夹配置需要：主机、文件夹路径
if not config.get("host") or not config.get("folder_path"):
- raise HTTPException(status_code=400, detail="远程文件夹配置必须包含主机和文件夹路径")
+ raise HTTPException(status_code=400, detail="文件夹配置必须包含主机和文件夹路径")
# 替换路径中的特殊字符为下划线
folder_path = config["folder_path"].lower().replace("/", "_").replace(":", "_").replace("\\", "_")
# 去除开头的下划线
@@ -1453,20 +1387,13 @@ async def create_config(config: Dict[str, Any]):
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
+ elif config_type == 'folder':
+ # For 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"已存在相同服务器和路径的远程文件夹配置。如需调整，请点击配置列表中的配置并修改配置内容。"
+ detail=f"已存在相同服务器和路径的文件夹配置。如需调整，请点击配置列表中的配置并修改配置内容。"
)
except sqlite3.OperationalError as e:
# 表不存在的情况，会在后面创建表
@@ -1498,7 +1425,7 @@ async def create_config(config: Dict[str, Any]):
global sync_manager
if sync_manager is not None:
# Create appropriate data source config object
- from config import BaseDataSourceConfig, DatabaseDataSourceConfig, LocalFolderDataSourceConfig, RemoteFolderDataSourceConfig
+ from config import BaseDataSourceConfig, DatabaseDataSourceConfig, FolderDataSourceConfig
if config_type == "database":
source_config = DatabaseDataSourceConfig(
@@ -1523,19 +1450,8 @@ async def create_config(config: Dict[str, Any]):
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(
+ elif config_type == "folder":
+ source_config = FolderDataSourceConfig(
name=config_id,
folder_path=config.get("folder_path"),
host=config.get("host"),
@@ -1570,36 +1486,7 @@ async def create_config(config: Dict[str, Any]):
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")
@@ -1613,9 +1500,9 @@ async def create_remote_folder_config(config: Dict[str, Any]):
Returns:
Created configuration with ID
"""
- # Set type to remote_folder if not provided
+ # Set type to folder if not provided
if "type" not in config:
- config["type"] = "remote_folder"
+ config["type"] = "folder"
# Call the generic create_config function
return await create_config(config)
@@ -1660,29 +1547,22 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]):
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
+ elif config_type == "folder":
+ # 文件夹配置：根据是否有host字段区分本地和远程
if not config.get("folder_path"):
- raise HTTPException(status_code=400, detail="Folder path is required for local folder configuration")
+ raise HTTPException(status_code=400, detail="Folder path is required for 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}"
+ # 根据是否有host字段生成不同的ID
+ if config.get("host") and config["host"] != "localhost":
+ # 远程文件夹：使用主机和文件夹路径生成ID
+ new_config_id = f"{config_type}_{config['host'].lower()}_{folder_path}"
+ else:
+ # 本地文件夹：使用文件夹路径生成ID
+ new_config_id = f"{config_type}_{folder_path}"
# 如果无法生成新的有意义的ID，保留原来的ID
if not new_config_id:
@@ -1708,7 +1588,7 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]):
global sync_manager
if sync_manager is not None:
# Create appropriate data source config object
- from config import BaseDataSourceConfig, DatabaseDataSourceConfig, LocalFolderDataSourceConfig, RemoteFolderDataSourceConfig
+ from config import BaseDataSourceConfig, DatabaseDataSourceConfig, FolderDataSourceConfig
if config_type == "database":
source_config = DatabaseDataSourceConfig(
@@ -1733,19 +1613,8 @@ async def update_folder_config(config_id: str, config: Dict[str, Any]):
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(
+ elif config_type == "folder":
+ source_config = FolderDataSourceConfig(
name=new_config_id,
folder_path=config.get("folder_path"),
host=config.get("host"),
diff --git a/config.py b/config.py
index d4a286c..238b17f 100644
--- a/config.py
+++ b/config.py
@@ -6,6 +6,10 @@ import os
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Optional, List, Dict, Any
from pydantic import Field
+from loguru import logger
+
+
+
class BaseDataSourceConfig:
@@ -78,8 +82,8 @@ class DatabaseDataSourceConfig(BaseDataSourceConfig):
self.scp_password = scp_password
-class LocalFolderDataSourceConfig(BaseDataSourceConfig):
- """Local folder data source configuration"""
+class FolderDataSourceConfig(BaseDataSourceConfig):
+ """Folder data source configuration (local or remote via SSH/SFTP)"""
def __init__(
self,
name: str,
@@ -91,8 +95,8 @@ class LocalFolderDataSourceConfig(BaseDataSourceConfig):
recursive: bool = True,
ignore_patterns: Optional[List[str]] = None
):
- super().__init__(name, "local_folder")
- self.folder_path = folder_path # 本地文件夹路径
+ super().__init__(name, "folder")
+ self.folder_path = folder_path # 文件夹路径
self.host = host # 主机地址
self.port = port # 主机端口
self.username = username # 主机用户名
@@ -101,29 +105,6 @@ class LocalFolderDataSourceConfig(BaseDataSourceConfig):
self.ignore_patterns = ignore_patterns # 忽略的文件模式列表
-class RemoteFolderDataSourceConfig(BaseDataSourceConfig):
- """Remote folder data source configuration (via SCP)"""
- def __init__(
- self,
- name: str,
- folder_path: str,
- host: str,
- port: int,
- username: str,
- password: Optional[str] = None,
- recursive: bool = True,
- ignore_patterns: Optional[List[str]] = None
- ):
- super().__init__(name, "remote_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 # 忽略的文件模式列表
-
-
class Settings(BaseSettings):
"""
Application settings
@@ -258,9 +239,9 @@ class Settings(BaseSettings):
scp_username=ds_config.get('scp_username', None),
scp_password=ds_config.get('scp_password', None)
))
- elif source_type == 'local_folder':
- # Create local folder data source
- configs.append(LocalFolderDataSourceConfig(
+ elif source_type == 'folder':
+ # Create folder data source
+ configs.append(FolderDataSourceConfig(
name=name, # 使用数据库表中的name列
folder_path=ds_config.get('folder_path', '.'),
host=ds_config.get('host', 'localhost'),
@@ -270,18 +251,6 @@ class Settings(BaseSettings):
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=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),
- 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")
diff --git a/db_utils.py b/db_utils.py
new file mode 100644
index 0000000..a8132e8
--- /dev/null
+++ b/db_utils.py
@@ -0,0 +1,129 @@
+"""
+Database utilities for RAG system
+"""
+import sqlite3
+from pathlib import Path
+from datetime import datetime
+from typing import Tuple, Optional
+from loguru import logger
+
+
+def get_db_connection() -> Tuple[sqlite3.Connection, sqlite3.Cursor]:
+ """
+ Get a SQLite database connection with data_sources table initialized
+
+ Returns:
+ tuple: (connection, cursor)
+ """
+ DATA_DIR = Path(__file__).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
+
+ return conn, cursor
+
+
+def get_data_source_update_at(source_name: str) -> Optional[datetime]:
+ """
+ Get update_at for a data source from data_sources table
+
+ Args:
+ source_name: Name of the data source
+
+ Returns:
+ datetime: Update time if found, None otherwise
+ """
+ try:
+ conn, cursor = get_db_connection()
+ try:
+ cursor.execute('SELECT update_at FROM data_sources WHERE name = ?', (source_name,))
+ result = cursor.fetchone()
+ if result and result[0]:
+ return datetime.fromisoformat(result[0])
+ return None
+ finally:
+ conn.close()
+ except Exception as e:
+ logger.warning(f"Error reading update_at from data_sources: {e}")
+ return None
+
+
+def update_data_source_update_at(source_name: str, update_at: datetime) -> bool:
+ """
+ Update update_at for a data source in data_sources table
+
+ Args:
+ source_name: Name of the data source
+ update_at: New update time
+
+ Returns:
+ bool: True if update succeeded, False otherwise
+ """
+ try:
+ conn, cursor = get_db_connection()
+ try:
+ cursor.execute('UPDATE data_sources SET update_at = ? WHERE name = ?', (update_at.isoformat(), source_name))
+ conn.commit()
+ logger.info(f"Updated update_at in data_sources for {source_name}: {update_at}")
+ return True
+ finally:
+ conn.close()
+ except Exception as e:
+ logger.warning(f"Error updating update_at in data_sources: {e}")
+ return False
+
+
+def init_session_db():
+ """
+ Initialize session database with users and sessions tables
+ """
+ DATA_DIR = Path(__file__).parent / "data"
+ DATA_DIR.mkdir(parents=True, exist_ok=True)
+ DB_PATH = DATA_DIR / "sessions.db"
+
+ conn = sqlite3.connect(DB_PATH)
+ try:
+ conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS users (
+ id TEXT PRIMARY KEY,
+ username TEXT UNIQUE,
+ password TEXT,
+ create_time TEXT
+ )
+ """
+ )
+ conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS sessions (
+ id TEXT PRIMARY KEY,
+ user_login TEXT,
+ title TEXT,
+ data TEXT,
+ update_time TEXT
+ )
+ """
+ )
+ # Improve concurrency for writes
+ conn.execute("PRAGMA journal_mode=WAL;")
+ conn.execute("PRAGMA synchronous=NORMAL;")
+ conn.commit()
+ finally:
+ conn.close()
diff --git a/rag/__init__.py b/rag/__init__.py
index 4d6c46e..4549796 100644
--- a/rag/__init__.py
+++ b/rag/__init__.py
@@ -2,10 +2,9 @@
RAG module for retrieval and generation
"""
from .vector_store import VectorStoreManager
-from .document_processor import DocumentProcessor
from .rag_engine import RAGEngine
from .file_parser import FileParser
from .chunk_handler import OptimizedDeltaThinkFilter
-__all__ = ["VectorStoreManager", "DocumentProcessor", "RAGEngine", "FileParser", "OptimizedDeltaThinkFilter"]
+__all__ = ["VectorStoreManager", "RAGEngine", "FileParser", "OptimizedDeltaThinkFilter"]
diff --git a/rag/document_processor.py b/rag/document_processor.py
deleted file mode 100644
index 929e0dd..0000000
--- a/rag/document_processor.py
+++ /dev/null
@@ -1,202 +0,0 @@
-"""
-Document processing and chunking module
-"""
-from llama_index.core import Document
-from llama_index.core.node_parser import SentenceSplitter
-from typing import List, Dict, Optional
-from datetime import datetime, date
-from loguru import logger
-from config import settings
-
-
-class DocumentProcessor:
- """Process and chunk documents for RAG"""
-
- def __init__(self):
- self.node_parser = SentenceSplitter(
- chunk_size=settings.CHUNK_SIZE,
- chunk_overlap=settings.CHUNK_OVERLAP
- )
-
- def mysql_doc_to_llamaindex_doc(self, mysql_doc: Dict, db_config=None) -> Document:
- """
- Convert MySQL document to LlamaIndex Document
- 将单个MySQL文档转换为LlamaIndex文档对象
-
- Args:
- mysql_doc: Dictionary from MySQL query result
- db_config: DatabaseConfig object (optional, for backward compatibility)
-
- Returns:
- LlamaIndex Document object
- """
- # Get column names from db_config or fallback to default values
- id_column = db_config.id_column if db_config else "id"
- title_column = db_config.title_column if db_config else "title"
- metadata_columns = db_config.metadata_columns if db_config else None
-
- doc_id = str(mysql_doc.get(id_column, ""))
-
- # 处理多个 content 列（支持合并多个列的内容）
- if db_config:
- # 使用配置的多个 content 列
- content_columns = db_config.content_columns
- content_separator = db_config.content_separator
- else:
- # 向后兼容：使用单个 content_column
- content_columns = ["content"] # Default to "content" column
- content_separator = "\n"
-
- # 合并所有 content 列的内容
- content_parts = []
- for col in content_columns:
- col_value = mysql_doc.get(col, "")
- if col_value:
- content_parts.append(str(col_value))
-
- # 用指定的分隔符连接多个列的内容
- content = content_separator.join(content_parts) if content_parts else ""
-
- title = mysql_doc.get(title_column, "") if title_column else None
-
- # Create unique doc_id with database source and table name to avoid conflicts
- # Format: {db_source}_{table_name}_{id} to ensure uniqueness across multiple tables
- # This must match the format used in sync_service.py
- if '_db_source' in mysql_doc and '_db_table' in mysql_doc:
- unique_doc_id = f"{mysql_doc['_db_source']}_{mysql_doc['_db_table']}_{doc_id}"
- elif '_db_source' in mysql_doc:
- unique_doc_id = f"{mysql_doc['_db_source']}_{doc_id}"
- else:
- unique_doc_id = doc_id
-
- # Build metadata
- # Note: metadata['doc_id'] should use unique_doc_id to match sync_service.py logic
- # Format: {db_source}_{table_name}_{id} to ensure uniqueness across multiple tables
- metadata = {
- "doc_id": unique_doc_id, # Format: {db_source}_{table_name}_{id}
- "original_doc_id": doc_id, # Keep original ID for reference
- "source": "mysql",
- "content_column": db_config.content_column # 额外存入 content_column
- }
-
- # Add database source information if available
- if '_db_source' in mysql_doc:
- metadata['db_source'] = mysql_doc['_db_source']
- if '_db_database' in mysql_doc:
- metadata['db_database'] = mysql_doc['_db_database']
- if '_db_table' in mysql_doc:
- metadata['db_table'] = mysql_doc['_db_table']
-
- if title:
- metadata["title"] = title
-
- # Add additional metadata columns if specified
- if metadata_columns:
- metadata_cols = [col.strip() for col in metadata_columns.split(",")]
- for col in metadata_cols:
- if col in mysql_doc and not col.startswith('_'):
- value = mysql_doc[col]
- # Convert datetime/date objects to strings for ChromaDB compatibility
- # ChromaDB metadata only supports: str, int, float, bool
- if isinstance(value, (datetime, date)):
- # Format datetime/date as ISO format string
- metadata[col] = value.isoformat()
- elif value is not None:
- # Convert other types to string if not already a supported type
- if not isinstance(value, (str, int, float, bool)):
- metadata[col] = str(value)
- else:
- metadata[col] = value
-
- # Create Document
- # Note: LlamaIndex uses id_ (not doc_id) as the unique identifier for Document
- # See: https://docs.llamaindex.org.cn/en/stable/module_guides/indexing/document_management/#update
- doc = Document(
- text=content,
- id_=unique_doc_id, # Use id_ as per LlamaIndex documentation
- metadata=metadata
- )
-
- return doc
-
- def process_documents(self, mysql_docs: List[Dict], db_config=None) -> List[Document]:
- """
- Process multiple MySQL documents into LlamaIndex Documents
- 处理多个MySQL文档，将其转换为LlamaIndex文档对象
- Args:
- mysql_docs: List of MySQL document dictionaries
- db_config: DatabaseConfig object (optional, for backward compatibility)
-
- Returns:
- List of LlamaIndex Document objects
- """
- documents = []
- id_column = db_config.id_column if db_config else "id"
- for mysql_doc in mysql_docs:
- try:
- doc = self.mysql_doc_to_llamaindex_doc(mysql_doc, db_config)
- if len(doc.text.strip()) >= 100:
- documents.append(doc)
- else:
- logger.warning(f"跳过过短文档 (id: {doc.id_})，内容长度: {len(doc.text)} 字符")
- except Exception as e:
- doc_id = mysql_doc.get(id_column, mysql_doc.get('_db_source', 'unknown'))
- logger.error(f"Error processing document {doc_id}: {e}")
- continue
-
- logger.info(f"Processed {len(documents)} documents")
- return documents
-
- def chunk_documents(self, documents: List[Document]) -> List[Document]:
- """
- Chunk documents into smaller pieces
-
- Args:
- documents: List of Document objects
- doc = Document(
- text=content,
- id_=unique_doc_id, # Use id_ as per LlamaIndex documentation
- metadata=metadata
- )
-
- Returns:
- List of chunked Document objects
- """
- chunked_docs = []
- for doc in documents:
- try:
- nodes = self.node_parser.get_nodes_from_documents([doc])
- # Get unique_doc_id from doc.id_ or metadata['doc_id']
- # This is the original document's unique ID (format: {db_source}_{table_name}_{id})
- unique_doc_id = getattr(doc, 'id_', None) or (doc.metadata.get('doc_id') if doc.metadata else None)
- if not unique_doc_id:
- # Fallback: use node.node_id if unique_doc_id is not available
- unique_doc_id = f"doc_{id(doc)}"
- logger.warning(f"Document has no id_ or doc_id in metadata, using fallback: {unique_doc_id}")
-
- # Convert nodes back to documents for storage
- # Use unique_doc_id + chunk_index as id_ for easier identification and duplicate checking
- # Format: {db_source}_{table_name}_{id}_chunk_{index}
- for index, node in enumerate(nodes):
- chunk_id = f"{unique_doc_id}_chunk_{index}"
- chunked_doc = Document(
- text=node.text,
- id_=chunk_id, # Use unique_doc_id + chunk_index for better traceability
- metadata={
- **doc.metadata,
- "chunk_id": chunk_id,
- "chunk_index": index,
- "total_chunks": len(nodes),
- "original_node_id": node.node_id # Keep original node_id for reference
- }
- )
- chunked_docs.append(chunked_doc)
- except Exception as e:
- # Access id_ property (not doc_id) for error logging
- doc_id = getattr(doc, 'id_', getattr(doc, 'doc_id', 'unknown'))
- logger.error(f"Error chunking document {doc_id}: {e}")
- continue
-
- logger.info(f"Chunked {len(documents)} documents into {len(chunked_docs)} chunks")
- return chunked_docs
-
diff --git a/rag/file_parser.py b/rag/file_parser.py
index 7ff858b..c55956f 100644
--- a/rag/file_parser.py
+++ b/rag/file_parser.py
@@ -3,6 +3,7 @@ File parsing module for various document formats
"""
import os
import requests
+from config import settings
import mimetypes
from typing import List, Dict, Optional
from pathlib import Path
@@ -42,74 +43,46 @@ class FileParser:
ext = Path(filename).suffix.lower() # 获取文件类型并转为小写
return ext in self.SUPPORTED_EXTENSIONS
- def _generate_doc_id(self, filename: str, host: Optional[str] = None) -> str:
+ def parse_file_content(self, content: bytes, file_path: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None, host: Optional[str] = None) -> List[Document]:
"""
- Generate a unique doc_id based on host address and filename
+ Parse file content from bytes into LlamaIndex Documents
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 and host)
+ content: File content as bytes
+ file_path: Original file path (for format detection and metadata)
+ doc_id: Optional document ID
metadata: Optional metadata to add to documents
- host: Optional host address (if not provided, will use local machine IP)
+ host: Optional host address (for remote files, use the remote host address)
Returns:
List of LlamaIndex Document objects
"""
- if not os.path.exists(file_path):
- raise FileNotFoundError(f"File not found: {file_path}")
+ import tempfile
- filename = os.path.basename(file_path)
- ext = Path(file_path).suffix.lower()
+ # Extract filename from path
+ original_filename = Path(file_path).name
+ ext = Path(file_path).suffix.lower() # 确定文件扩展名
+ if not ext:
+ # 无拓展名时，Try to detect from mimetype
+ mime_type, _ = mimetypes.guess_type(file_path)
+ if mime_type:
+ ext = mimetypes.guess_extension(mime_type) or '.txt'
+ else:
+ ext = '.txt'
+
+ # Check if file format is supported
if not self.is_supported(file_path):
raise ValueError(f"Unsupported file format: {ext}. Supported formats: {', '.join(self.SUPPORTED_EXTENSIONS)}")
- # 生成基于主机地址和真实文件名的doc_id（如果未提供）
- if not doc_id:
- doc_id = self._generate_doc_id(filename, host)
-
try:
- logger.info(f"Starting to parse file: {filename} (type: {ext})\n")
+ logger.info(f"Starting to parse file: {original_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')}
+ files = {'file': (original_filename, content, 'application/msword')}
soffice_url = f"http://{settings.SOFFICE_HOST}:{settings.SOFFICE_PORT}/convert"
response = requests.post(soffice_url, files=files, timeout=60)
try:
@@ -118,17 +91,14 @@ class FileParser:
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
+ parse_path = tmp_docx_file.name
else:
- # 对于其他文件格式，直接使用原始文件路径
- parse_path = file_path
- cleanup_tmp = False
+ # 对于其他文件格式，创建临时文件保存内容
+ with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp_file:
+ tmp_file.write(content)
+ parse_path = tmp_file.name
# Use LlamaIndex's SimpleDirectoryReader for parsing
# It supports many formats out of the box
@@ -144,27 +114,18 @@ class FileParser:
# Validate that we got documents
if not documents:
- logger.warning(f"No documents extracted from file: {filename}")
- raise ValueError(f"Failed to extract content from {filename}. The file may be empty or in an unsupported format.")
-
- # Check if documents have content
- empty_docs = []
- for i, doc in enumerate(documents):
- if not doc.text or not doc.text.strip():
- empty_docs.append(i)
- logger.warning(f"Document {i} from {filename} has empty content")
-
- if len(empty_docs) == len(documents):
- raise ValueError(f"All documents from {filename} have empty content. The file may not contain readable text.")
+ logger.warning(f"No documents extracted from file: {original_filename}")
+ raise ValueError(f"Failed to extract content from {original_filename}. The file may be empty or in an unsupported format.")
# Add metadata to each document
base_metadata = {
'source': 'file_upload',
- 'file_name': filename,
- 'file_path': file_path,
'file_type': ext.lstrip('.'),
+ 'file_name': original_filename,
+ 'file_path': file_path, # Use the original file path
}
+ # Update with provided metadata if any
if metadata:
base_metadata.update(metadata)
@@ -173,7 +134,7 @@ class FileParser:
for i, doc in enumerate(documents):
# Skip empty documents
if not doc.text or not doc.text.strip():
- logger.warning(f"Skipping empty document {i} from {filename}")
+ logger.warning(f"Skipping empty document {i} from {original_filename}")
continue
# Set document ID
@@ -189,70 +150,18 @@ class FileParser:
valid_documents.append(doc)
- 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)})\n")
+ logger.info(f"Successfully parsed file {original_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, host: Optional[str] = None) -> List[Document]:
- """
- Parse file content from bytes into LlamaIndex Documents
-
- Args:
- content: File content as bytes
- 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
- """
- import tempfile
-
- # Create temporary file
- ext = Path(filename).suffix.lower() # 确定文件扩展名
- if not ext:
- # 无拓展名时，Try to detect from mimetype
- mime_type, _ = mimetypes.guess_type(filename)
- if mime_type:
- ext = mimetypes.guess_extension(mime_type) or '.txt'
- 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)
- tmp_path = tmp_file.name
-
- try:
- # Parse the temporary file
- documents = self.parse_file(tmp_path, doc_id=doc_id, metadata=metadata)
- return documents
- finally:
- # Clean up temporary file
try:
- os.unlink(tmp_path)
+ os.unlink(parse_path)
+ logger.debug(f"Cleaned up temporary file: {parse_path}")
except Exception as e:
- logger.warning(f"Failed to delete temporary file {tmp_path}: {e}")
+ logger.warning(f"Failed to delete temporary file {parse_path}: {e}")
diff --git a/rag/vector_store.py b/rag/vector_store.py
index b98b24d..60ac5a3 100644
--- a/rag/vector_store.py
+++ b/rag/vector_store.py
@@ -386,7 +386,6 @@ class VectorStoreManager:
skipped_count = 0
if skip_existing:
- existing_ids = self.get_existing_doc_ids()
for doc in documents:
# Get document id_ (for chunks, this is {unique_doc_id}_chunk_{index})
# For non-chunk documents, this is {unique_doc_id}
@@ -395,7 +394,7 @@ class VectorStoreManager:
if not doc_id_attr:
# Document has no id_, add it (will be processed)
new_documents.append(doc)
- elif doc_id_attr in existing_ids:
+ elif self.document_exists(doc_id_attr):
# This specific chunk/document already exists, skip it
skipped_count += 1
else:
diff --git a/static/config/index.html b/static/config/index.html
index 803848a..0ec14d1 100644
--- a/static/config/index.html
+++ b/static/config/index.html
@@ -4,8 +4,8 @@
RAG 配置管理
-
-
+
+
@@ -18,8 +18,7 @@
@@ -28,9 +27,6 @@
-
@@ -78,8 +74,7 @@
`;
formElement.appendChild(basicSection);
// 根据配置类型生成相应的表单字段
- if (config.type === 'local_folder' || config.type === 'remote_folder') {
+ if (config.type === 'folder') {
const folderSection = document.createElement('div');
folderSection.innerHTML = `
文件夹配置
@@ -257,8 +250,8 @@ function generateConfigForm(config) {
}
- // 为local_folder和remote_folder添加SSH连接配置
- if (config.type === 'remote_folder' || config.type === 'local_folder') {
+ // 为folder添加SSH连接配置
+ if (config.type === 'folder') {
const sshSection = document.createElement('div');
sshSection.innerHTML = `
SSH连接配置
@@ -1042,14 +1035,7 @@ async function handleAddConfigDirectly() {
id_column: '',
content_column: ''
};
- } else if (configType === 'local_folder') {
- tempConfig = {
- type: configType,
- folder_path: '',
- recursive: true,
- ignore_patterns: []
- };
- } else if (configType === 'remote_folder') {
+ } else if (configType === 'folder') {
tempConfig = {
type: configType,
folder_path: '',
@@ -1108,15 +1094,7 @@ async function handleAddConfig(event) {
id_column: '',
content_column: ''
};
- } else if (configType === 'local_folder') {
- configData = {
- name: configName,
- type: configType,
- folder_path: '',
- recursive: true,
- ignore_patterns: []
- };
- } else if (configType === 'remote_folder') {
+ } else if (configType === 'folder') {
configData = {
name: configName,
type: configType,
@@ -1191,17 +1169,14 @@ async function saveConfig() {
if (formData.type === 'database') {
// 数据库配置：database_数据库名_表名
generatedName = `database_${formData.database || 'unknown'}_${formData.table_name || 'unknown'}`;
- } else if (formData.type === 'local_folder') {
- // 本地文件夹：local_folder_文件夹路径（替换特殊字符）
+ } else if (formData.type === 'folder') {
+ // 文件夹：folder_主机_文件夹路径（替换特殊字符）
const folderName = formData.folder_path ? formData.folder_path.replace(/[\\/:*?"<>|]/g, '_') : 'unknown';
- generatedName = `local_folder_${folderName}`;
- } else if (formData.type === 'remote_folder') {
- // 远程文件夹：remote_folder_主机_文件夹路径（替换特殊字符）
- const folderName = formData.folder_path ? formData.folder_path.replace(/[\\/:*?"<>|]/g, '_') : 'unknown';
- generatedName = `remote_folder_${formData.host || 'unknown'}_${folderName}`;
+ generatedName = `folder_${formData.host || 'unknown'}_${folderName}`;
} else {
- // 默认名称
- generatedName = `default_config`;
+ // 不支持的配置类型
+ alert('不支持的配置类型');
+ return;
}
}
@@ -1218,11 +1193,8 @@ async function saveConfig() {
if (!formData.content_column || formData.content_column.trim() === '') {
missingFields.push('内容列');
}
- } else if (formData.type === 'local_folder' || formData.type === 'remote_folder') {
+ } else if (formData.type === 'folder') {
if (!formData.folder_path) missingFields.push('文件夹路径');
- }
-
- if (formData.type === 'remote_folder' || formData.type === 'local_folder') {
if (!formData.host) missingFields.push('主机地址');
if (!formData.port) missingFields.push('端口');
if (!formData.username) missingFields.push('用户名');
@@ -1268,6 +1240,9 @@ async function saveConfig() {
// 获取保存/更新后的配置
const updatedConfig = await response.json();
+ // 检查是否是新配置（POST）
+ const isNewConfig = currentConfig.id === null;
+
// 更新当前配置的ID
currentConfig.id = updatedConfig.id;
@@ -1282,7 +1257,7 @@ async function saveConfig() {
// 数据源名称是配置的ID，而不是config中的name字段
const sourceName = updatedConfig.id;
- if (currentConfig.id === null) {
+ if (isNewConfig) {
// 新配置（POST）：保存成功后，立刻启动数据源的同步操作
alert('配置保存成功，正在启动同步...');
@@ -1359,21 +1334,18 @@ function collectFormData() {
formData.type = document.getElementById('formType').value;
// 文件夹配置
- if (formData.type === 'local_folder' || formData.type === 'remote_folder') {
+ if (formData.type === 'folder') {
formData.folder_path = document.getElementById('formFolderPath').value;
formData.recursive = document.getElementById('formRecursive').checked;
const ignorePatterns = document.getElementById('formIgnorePatterns').value;
formData.ignore_patterns = ignorePatterns ? ignorePatterns.split(',').map(p => p.trim()) : [];
- }
-
- // 远程连接配置和本地文件夹的SSH配置
- if (formData.type === 'remote_folder' || formData.type === 'local_folder') {
+
+ // 文件夹的SSH配置
formData.host = document.getElementById('formHost').value;
formData.port = parseInt(document.getElementById('formPort').value);
formData.username = document.getElementById('formUsername').value;
formData.password = document.getElementById('formPassword').value;
-
}
// 数据库配置
@@ -1396,19 +1368,7 @@ function collectFormData() {
formData.mysql_user = document.getElementById('formMysqlUser').value;
formData.mysql_password = document.getElementById('formMysqlPassword').value;
- // 仅当配置类型不是database时，才处理文件列
- if (formData.type !== 'database') {
- formData.file_column = document.getElementById('formFileColumn').value;
-
- // 确保文件列同时出现在内容列中
- if (formData.file_column && formData.file_column !== '') {
- const contentColumns = formData.content_column ? formData.content_column.split(',') : [];
- if (!contentColumns.includes(formData.file_column)) {
- contentColumns.push(formData.file_column);
- formData.content_column = contentColumns.join(',');
- }
- }
- }
+
}
diff --git a/static/config/style.css b/static/config/style.css
index 85f109a..b23a011 100644
--- a/static/config/style.css
+++ b/static/config/style.css
@@ -192,16 +192,11 @@ body {
color: #1976d2;
}
-.config-item[data-type="local_folder"] .config-item-icon {
+.config-item[data-type="folder"] .config-item-icon {
background-color: #e8f5e9;
color: #388e3c;
}
-.config-item[data-type="remote_folder"] .config-item-icon {
- background-color: #fff3e0;
- color: #f57c00;
-}
-
.config-item-details {
flex: 1;
}
diff --git a/sync/__init__.py b/sync/__init__.py
index 7c6c88a..305bf76 100644
--- a/sync/__init__.py
+++ b/sync/__init__.py
@@ -1,13 +1,11 @@
"""Synchronization modules for all data sources"""
from .base_sync import BaseSync, get_sync_class
from .mysql_sync import MySQLSync
-from .local_folder_sync import LocalFolderSync
-from .remote_folder_sync import RemoteFolderSync
+from .folder_sync import FolderSync
__all__ = [
'BaseSync',
'get_sync_class',
'MySQLSync',
- 'LocalFolderSync',
- 'RemoteFolderSync'
+ 'FolderSync'
]
diff --git a/sync/base_sync.py b/sync/base_sync.py
index 31541ef..7a9c9a5 100644
--- a/sync/base_sync.py
+++ b/sync/base_sync.py
@@ -8,15 +8,17 @@ from rag.file_parser import FileParser
class BaseSync(ABC):
"""Abstract base class for all data source syncers"""
- def __init__(self, config: BaseDataSourceConfig):
+ def __init__(self, config: BaseDataSourceConfig, vector_store_manager=None):
"""
Initialize sync with data source configuration
Args:
config: Configuration for the data source
+ vector_store_manager: Vector store manager for document existence checks
"""
self.config = config
self.file_parser = FileParser()
+ self.vector_store_manager = vector_store_manager
def _get_file_bytes(self, file_path: str, source_type: str, source_config: Optional[Dict[str, Any]] = None) -> Optional[bytes]:
"""
@@ -89,33 +91,7 @@ class BaseSync(ABC):
logger.error(f"Error getting file bytes for {file_path}: {e}")
return None
- def _parse_file_content(self, file_bytes: bytes, file_path: str, host: Optional[str] = None) -> Optional[str]:
- """
- Parse file content using FileParser class
-
- Args:
- file_bytes: File content as bytes
- file_path: Path to the file (for getting file extension)
- host: Optional host address (for remote files, use the remote host address)
-
- Returns:
- Parsed file content if successfully parsed, None otherwise
- """
- try:
- # 使用 FileParser 解析文件内容
- # parse_file_content 方法的参数顺序是 content, filename
- documents = self.file_parser.parse_file_content(file_bytes, file_path, host=host)
- if documents:
- # 合并所有文档内容
- return '\n\n'.join(doc.text for doc in documents if doc.text)
- else:
- from loguru import logger
- logger.warning(f"No content extracted from {file_path}")
- return None
- except Exception as e:
- from loguru import logger
- logger.error(f"Error parsing file content for {file_path}: {e}")
- return None
+
@abstractmethod
def fetch_all_documents(self) -> List[Dict[str, Any]]:
@@ -128,13 +104,121 @@ class BaseSync(ABC):
pass
@abstractmethod
- def fetch_new_documents(self, last_sync_time=None, synced_doc_ids=None) -> List[Dict[str, Any]]:
+ def doc_to_llamaindex_doc(self, doc: Dict) -> 'Document':
+ """
+ Convert data source document to LlamaIndex Document
+
+ Args:
+ doc: Document from the data source
+
+ Returns:
+ LlamaIndex Document object
+ """
+ pass
+
+ def process_documents(self, docs: List[Dict]) -> List['Document']:
+ """
+ Process multiple documents into LlamaIndex Documents
+
+ Args:
+ docs: List of documents from the data source
+
+ Returns:
+ List of LlamaIndex Document objects
+ """
+ from loguru import logger
+ documents = []
+ for doc in docs:
+ try:
+ llamaindex_doc = self.doc_to_llamaindex_doc(doc)
+ if len(llamaindex_doc.text.strip()) >= 1:
+ documents.append(llamaindex_doc)
+ else:
+ logger.warning(f"跳过过短文档 (id: {llamaindex_doc.id_})，内容长度: {len(llamaindex_doc.text)} 字符")
+ except Exception as e:
+ doc_id = doc.get('id', 'unknown')
+ logger.error(f"Error processing document {doc_id}: {e}")
+ continue
+
+ logger.info(f"Processed {len(documents)} documents")
+ return documents
+
+ @staticmethod
+ def chunk_documents(documents: List['Document']) -> List['Document']:
+ """
+ Chunk documents into smaller pieces
+
+ Args:
+ documents: List of Document objects to chunk
+
+ Returns:
+ List of chunked Document objects
+ """
+ from llama_index.core import Document
+ from llama_index.core.node_parser import SentenceSplitter
+ from loguru import logger
+ from config import settings
+
+ node_parser = SentenceSplitter(
+ chunk_size=settings.CHUNK_SIZE,
+ chunk_overlap=settings.CHUNK_OVERLAP
+ )
+
+ chunked_docs = []
+ for doc in documents:
+ try:
+ nodes = node_parser.get_nodes_from_documents([doc])
+ # Get unique_doc_id from doc.id_ or metadata['doc_id']
+ unique_doc_id = getattr(doc, 'id_', None) or (doc.metadata.get('doc_id') if doc.metadata else None)
+ if not unique_doc_id:
+ # Fallback: use node.node_id if unique_doc_id is not available
+ unique_doc_id = f"doc_{id(doc)}"
+ logger.warning(f"Document has no id_ or doc_id in metadata, using fallback: {unique_doc_id}")
+
+ # Convert nodes back to documents for storage
+ for index, node in enumerate(nodes):
+ chunk_id = f"{unique_doc_id}_chunk_{index}"
+ chunked_doc = Document(
+ text=node.text,
+ id_=chunk_id,
+ metadata={
+ **doc.metadata,
+ "chunk_id": chunk_id,
+ "chunk_index": index,
+ "total_chunks": len(nodes),
+ "original_node_id": node.node_id
+ }
+ )
+ chunked_docs.append(chunked_doc)
+ except Exception as e:
+ # Access id_ property (not doc_id) for error logging
+ doc_id = getattr(doc, 'id_', getattr(doc, 'doc_id', 'unknown'))
+ logger.error(f"Error chunking document {doc_id}: {e}")
+ continue
+
+ logger.info(f"Chunked {len(documents)} documents into {len(chunked_docs)} chunks")
+ return chunked_docs
+
+ @abstractmethod
+ def generate_doc_id(self, identifier: str) -> str:
+ """
+ Generate a unique document ID for different data sources
+
+ Args:
+ identifier: Unique identifier for the document (file path, record ID, etc.)
+
+ Returns:
+ Unique document ID
+ """
+ pass
+
+ @abstractmethod
+ def fetch_new_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
"""
Fetch new/updated documents from the data source since last sync time
Args:
last_sync_time: Last synchronization time
- synced_doc_ids: Set of document IDs that have already been synced
Returns:
List of new/updated documents
@@ -192,7 +276,7 @@ def get_sync_class(source_type: str) -> type[BaseSync]:
Get the appropriate sync class based on data source type
Args:
- source_type: Type of data source (database, local_folder, remote_folder)
+ source_type: Type of data source (database, folder)
Returns:
Sync class corresponding to the data source type
@@ -201,13 +285,11 @@ def get_sync_class(source_type: str) -> type[BaseSync]:
ValueError: If source type is not supported
"""
from sync.mysql_sync import MySQLSync
- from sync.local_folder_sync import LocalFolderSync
- from sync.remote_folder_sync import RemoteFolderSync
+ from sync.folder_sync import FolderSync
sync_classes = {
'database': MySQLSync, # Currently only MySQL, but can be extended
- 'local_folder': LocalFolderSync,
- 'remote_folder': RemoteFolderSync
+ 'folder': FolderSync
}
if source_type not in sync_classes:
diff --git a/sync/folder_sync.py b/sync/folder_sync.py
new file mode 100644
index 0000000..26505ab
--- /dev/null
+++ b/sync/folder_sync.py
@@ -0,0 +1,384 @@
+"""Folder synchronization implementation for local and remote folders"""
+import os
+import re
+from typing import List, Dict, Any, Set
+from datetime import datetime
+from pathlib import Path
+from loguru import logger
+from config import BaseDataSourceConfig
+from sync.base_sync import BaseSync
+from rag.file_parser import FileParser
+from llama_index.core import Document
+
+
+class FolderSync(BaseSync):
+ """Handle synchronization between folder (local or remote) and ChromaDB"""
+
+ def __init__(self, config: BaseDataSourceConfig):
+ """
+ Initialize folder sync with configuration
+
+ Args:
+ config: Folder configuration
+ """
+ super().__init__(config)
+ self.file_parser = FileParser()
+ self._ssh_client = None
+ self._sftp_client = None
+
+ def fetch_all_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
+ """
+ Fetch documents from the folder
+
+ Args:
+ last_sync_time: Last synchronization time (for incremental sync)
+
+ Returns:
+ List of documents
+ """
+ documents = []
+ self._connect()
+ try:
+ files = self._get_all_files()
+
+ for file_path in files:
+
+ # Parse file content
+ try:
+ # 检查文件扩展名是否在支持的列表中
+ if Path(file_path).suffix.lower() not in FileParser.SUPPORTED_EXTENSIONS:
+ logger.debug(f"Skipping unsupported file: {file_path}")
+ continue
+
+ # Generate document ID
+ doc_id = self.generate_doc_id(file_path)
+
+ if last_sync_time is not None:
+ # Check if document has already been synced
+ if self.vector_store_manager and self.vector_store_manager.document_exists(doc_id):
+ # If document is already synced, check if it's been modified since last sync
+ if last_sync_time:
+ file_stat = self._sftp_client.stat(file_path)
+ file_mtime = datetime.fromtimestamp(file_stat.st_mtime)
+ # Skip if not modified since last sync
+ if file_mtime <= last_sync_time:
+ continue
+
+ # Read file content
+ with self._sftp_client.open(file_path, 'rb') as f:
+ file_bytes = f.read()
+
+ # Parse file content
+ if file_bytes:
+ try:
+ # 使用 FileParser 解析文件内容
+ parsed_docs = self.file_parser.parse_file_content(file_bytes, file_path, doc_id=doc_id, host=self.config.host)
+ if parsed_docs:
+ # 合并所有文档内容
+ content = '\n\n'.join(doc.text for doc in parsed_docs if doc.text)
+ else:
+ logger.warning(f"No content extracted from {file_path}")
+ content = f"[无法读取文件：{Path(file_path).name}]"
+ except Exception as e:
+ logger.error(f"Error parsing file content for {file_path}: {e}")
+ content = f"[无法读取文件：{Path(file_path).name}]"
+ else:
+ content = f"[无法读取文件：{Path(file_path).name}]"
+
+ # Build document
+ document = {
+ 'id': doc_id,
+ 'content': content,
+ 'metadata': {
+ 'file_path': str(file_path),
+ 'update_time': datetime.fromtimestamp(self._sftp_client.stat(file_path).st_mtime),
+ 'host': self.config.host
+ }
+ }
+ documents.append(document)
+ except Exception as e:
+ logger.error(f"Error processing file {file_path}: {e}")
+ finally:
+ self._disconnect()
+
+ return documents
+
+ def fetch_new_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
+ """
+ Fetch new/updated documents from the folder since last sync time
+
+ Args:
+ last_sync_time: Last synchronization time
+
+ Returns:
+ List of new/updated documents
+ """
+ # Call fetch_all_documents which now handles document existence checks
+ return self.fetch_all_documents(last_sync_time)
+
+ def get_synced_document_ids(self) -> Set[str]:
+ """
+ Get IDs of all files in the folder
+
+ Returns:
+ Set of file paths (as document IDs)
+ """
+ self._connect()
+ try:
+ files = self._get_all_files()
+ return set(files)
+ finally:
+ self._disconnect()
+
+ def generate_doc_id(self, file_path: str) -> str:
+ """
+ Generate a unique document ID for files
+
+ Args:
+ file_path: Path to the file
+
+ Returns:
+ Unique document ID based on host IP and file path
+ """
+ # 使用配置中的主机地址
+ host_address = self.config.host or 'unknown'
+
+ # 替换路径中的特殊字符，避免生成无效的doc_id
+ sanitized_path = file_path.replace('/', '_').replace('\\', '_').replace(':', '_').replace(' ', '_')
+ return f"{host_address}_{sanitized_path}"
+
+ def doc_to_llamaindex_doc(self, doc: Dict) -> 'Document':
+ """
+ Convert folder document to LlamaIndex Document
+
+ Args:
+ doc: Folder document dictionary
+
+ Returns:
+ LlamaIndex Document object
+ """
+ content = doc.get('content', "")
+ doc_id = doc.get('id', "")
+ metadata = doc.get('metadata', {})
+
+ # Ensure metadata has source information
+ metadata['source'] = 'folder'
+ metadata['host'] = self.config.host
+
+ # Create Document
+ return Document(
+ text=content,
+ id_=doc_id,
+ metadata=metadata
+ )
+
+ def _connect(self):
+ """
+ Connect to the server via SSH/SFTP
+
+ Raises:
+ Exception: If connection fails with detailed error message
+ """
+ import paramiko
+
+ self._ssh_client = paramiko.SSHClient()
+ self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+ # Connect to SSH server
+
+ # 获取用户名
+ username = self.config.username
+ if not username:
+ raise Exception("SSH connection failed: Username is required")
+
+ # 建立SSH连接的参数
+ ssh_params = {
+ 'hostname': self.config.host,
+ 'port': self.config.port or 22,
+ 'username': username,
+ 'password': self.config.password,
+ 'timeout': 10,
+ 'allow_agent': True, # 允许使用SSH代理
+ 'look_for_keys': False # 禁用查找本地密钥文件
+ }
+
+ try:
+ # 连接到SSH服务器
+ self._ssh_client.connect(**ssh_params)
+
+ # Create SFTP client
+ self._sftp_client = self._ssh_client.open_sftp()
+ except paramiko.AuthenticationException:
+ raise Exception(f"SSH connection failed: Authentication failed for user {username} on {self.config.host}")
+ except paramiko.SSHException as ssh_error:
+ raise Exception(f"SSH connection failed: {str(ssh_error)}")
+ except Exception as e:
+ raise Exception(f"Connection failed: {str(e)}")
+
+ def _disconnect(self):
+ """
+ Disconnect from the server
+ """
+ if self._sftp_client:
+ self._sftp_client.close()
+ self._sftp_client = None
+
+ if self._ssh_client:
+ self._ssh_client.close()
+ self._ssh_client = None
+
+ def _get_all_files(self) -> List[str]:
+ """
+ Get all files in the folder
+
+ Returns:
+ List of file paths
+
+ Note:
+ This method assumes that a connection has already been established by the caller
+ """
+ files = []
+ try:
+ # 直接调用 _get_files_recursive，使用已经建立的连接
+ self._get_files_recursive(self.config.folder_path, files)
+ except Exception as e:
+ logger.error(f"Error getting all files: {e}")
+ return files
+
+ def _get_files_recursive(self, folder_path: str, files: List[str]):
+ """
+ Recursively get all files in the folder
+
+ Args:
+ folder_path: Current folder path
+ files: List to store found files
+ """
+ try:
+ items = self._sftp_client.listdir_attr(folder_path)
+
+ for item in items:
+ item_path = os.path.join(folder_path, item.filename)
+
+ if item.filename not in ('.', '..'):
+ if item.st_mode & 0o040000: # Check if it's a directory
+ if self.config.recursive:
+ self._get_files_recursive(item_path, files)
+ else:
+ # Check if file should be ignored
+ if not self._should_ignore_file(item_path):
+ files.append(item_path)
+ except Exception as e:
+ logger.error(f"Error listing folder {folder_path}: {e}")
+
+ def _should_ignore_file(self, file_path: str) -> bool:
+ """
+ Check if file should be ignored based on ignore patterns
+
+ Args:
+ file_path: File path to check
+
+ Returns:
+ True if file should be ignored, False otherwise
+ """
+ if not hasattr(self.config, 'ignore_patterns') or not self.config.ignore_patterns:
+ return False
+
+ # Get relative path from folder root
+ relative_path = os.path.relpath(file_path, self.config.folder_path)
+
+ for pattern in self.config.ignore_patterns:
+ if self._match_pattern(relative_path, pattern):
+ return True
+
+ return False
+
+ def _match_pattern(self, path: str, pattern: str) -> bool:
+ """
+ Match a path against a pattern (similar to .gitignore)
+
+ Args:
+ path: Path to match
+ pattern: Pattern to match against
+
+ Returns:
+ True if path matches pattern, False otherwise
+ """
+ # Convert glob pattern to regex
+ regex_pattern = pattern
+ regex_pattern = regex_pattern.replace('.', r'\.')
+ regex_pattern = regex_pattern.replace('*', r'.*')
+ regex_pattern = regex_pattern.replace('?', r'.')
+
+ # Handle directory patterns
+ if pattern.endswith('/'):
+ regex_pattern = f'^{regex_pattern}.*$'
+ else:
+ regex_pattern = f'^{regex_pattern}$'
+
+ return bool(re.match(regex_pattern, path))
+
+ @staticmethod
+ def check_data_source_exists(config: BaseDataSourceConfig) -> bool:
+ """
+ Check if the folder exists and is accessible
+
+ Args:
+ config: Folder configuration
+
+ Returns:
+ True if folder exists and is accessible, False otherwise
+
+ Raises:
+ Exception: If connection fails with detailed error message
+ """
+ import paramiko
+
+ ssh_client = None
+ sftp_client = None
+ try:
+ # 检查必要的配置
+ if not config.host:
+ raise Exception("SSH connection failed: Host is required")
+
+ username = config.username
+ if not username:
+ raise Exception("SSH connection failed: Username is required")
+
+ ssh_client = paramiko.SSHClient()
+ ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+ # Connect to SSH server - use SSH agent if available, otherwise password
+ try:
+ ssh_client.connect(
+ hostname=config.host,
+ port=config.port or 22,
+ username=username,
+ password=config.password,
+ timeout=10,
+ allow_agent=True, # 允许使用SSH代理
+ look_for_keys=False # 禁用查找本地密钥文件
+ )
+ except paramiko.AuthenticationException:
+ raise Exception(f"SSH connection failed: Authentication failed for user {username} on {config.host}")
+ except paramiko.SSHException as ssh_error:
+ raise Exception(f"SSH connection failed: {str(ssh_error)}")
+ except Exception as e:
+ raise Exception(f"Connection failed: {str(e)}")
+
+ # Create SFTP client and check folder exists
+ try:
+ sftp_client = ssh_client.open_sftp()
+ sftp_client.stat(config.folder_path)
+ except Exception as e:
+ raise Exception(f"Folder access failed: {str(e)}")
+
+ return True
+ except Exception as e:
+ logger.error(f"Error checking folder: {e}")
+ # 重新抛出异常，以便上层能够捕获并传递详细的错误信息
+ raise
+ finally:
+ if sftp_client:
+ sftp_client.close()
+ if ssh_client:
+ ssh_client.close()
diff --git a/sync/local_folder_sync.py b/sync/local_folder_sync.py
deleted file mode 100644
index 9261f89..0000000
--- a/sync/local_folder_sync.py
+++ /dev/null
@@ -1,291 +0,0 @@
-"""Local folder synchronization implementation"""
-import os
-import re
-from typing import List, Dict, Any, Set
-from datetime import datetime
-from pathlib import Path
-from loguru import logger
-from config import BaseDataSourceConfig
-from sync.base_sync import BaseSync
-from rag.file_parser import FileParser
-
-
-class LocalFolderSync(BaseSync):
- """Handle synchronization between local folder and ChromaDB"""
-
- def __init__(self, config: BaseDataSourceConfig):
- """
- Initialize local folder sync with configuration
-
- Args:
- config: Local folder configuration
- """
- super().__init__(config)
- self.file_parser = FileParser()
- self._ssh_client = None
- self._sftp_client = None
-
- def fetch_all_documents(self) -> List[Dict[str, Any]]:
- """
- Fetch all documents from the local folder
-
- Returns:
- List of documents
- """
- return self._fetch_documents()
-
- def fetch_new_documents(self, last_sync_time=None, synced_doc_ids=None) -> List[Dict[str, Any]]:
- """
- Fetch new/updated documents from the local folder since last sync time
-
- Args:
- last_sync_time: Last synchronization time
- synced_doc_ids: Set of document IDs that have already been synced
-
- Returns:
- List of new/updated documents
- """
- return self._fetch_documents(last_sync_time, synced_doc_ids)
-
- def get_synced_document_ids(self) -> Set[str]:
- """
- Get IDs of all files in the local folder via SFTP
-
- Returns:
- Set of file paths (as document IDs)
- """
- self._connect()
- try:
- files = self._get_all_files()
- return set(files)
- finally:
- self._disconnect()
-
- def _fetch_documents(self, last_sync_time=None, synced_doc_ids=None) -> List[Dict[str, Any]]:
- """
- Internal method to fetch documents from local folder via SFTP
-
- Args:
- last_sync_time: Last synchronization time (for incremental sync)
- synced_doc_ids: Set of document IDs that have already been synced
-
- Returns:
- List of documents
- """
- documents = []
- self._connect()
- try:
- files = self._get_all_files()
-
- for file_path in files:
- # Check if document has already been synced
- if synced_doc_ids and file_path in synced_doc_ids:
- # If file is already synced, check if it's been modified since last sync
- if last_sync_time:
- file_stat = self._sftp_client.stat(file_path)
- file_mtime = datetime.fromtimestamp(file_stat.st_mtime)
- # Skip if not modified since last sync
- if file_mtime <= last_sync_time:
- continue
- # If file is not synced yet, always include it regardless of modification time
- # This handles the case where files were added to the folder after last_sync_time but have older mtimes
-
- # Parse file content
- try:
- # 检查文件扩展名是否在支持的列表中
- file_ext = os.path.splitext(file_path)[1].lower()
- if file_ext not in FileParser.SUPPORTED_EXTENSIONS:
- logger.debug(f"Skipping unsupported file: {file_path}")
- continue
-
- # 使用 SFTP 获取文件内容
- with self._sftp_client.open(file_path, 'rb') as f:
- file_bytes = f.read()
-
- # 使用 BaseSync 中的通用方法解析文件内容
- content = self._parse_file_content(file_bytes, file_path) if file_bytes else f"[无法读取文件：{os.path.basename(file_path)}]"
-
- file_stat = self._sftp_client.stat(file_path)
- document = {
- 'id': file_path,
- 'title': os.path.basename(file_path),
- 'content': content,
- 'file_path': file_path,
- 'update_time': datetime.fromtimestamp(file_stat.st_mtime)
- }
- documents.append(document)
- except Exception as e:
- logger.error(f"Error processing file {file_path}: {e}")
- finally:
- self._disconnect()
-
- return documents
-
- def _connect(self):
- """
- Connect to the host machine via SSH/SFTP
- """
- import paramiko
-
- self._ssh_client = paramiko.SSHClient()
- self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
- # Connect to SSH server
- # 确保username不为None，否则paramiko会报错
- username = self.config.username or ''
-
- # 建立SSH连接的参数
- ssh_params = {
- 'hostname': self.config.host,
- 'port': self.config.port or 22,
- 'username': username,
- 'password': self.config.password,
- 'timeout': 10,
- 'allow_agent': True, # 允许使用SSH代理
- 'look_for_keys': False # 禁用查找本地密钥文件
- }
-
- # 连接到SSH服务器
- self._ssh_client.connect(**ssh_params)
-
- # Create SFTP client
- self._sftp_client = self._ssh_client.open_sftp()
-
- def _disconnect(self):
- """
- Disconnect from the host machine
- """
- if self._sftp_client:
- self._sftp_client.close()
- self._sftp_client = None
-
- if self._ssh_client:
- self._ssh_client.close()
- self._ssh_client = None
-
- def _get_all_files(self) -> List[str]:
- """
- Get all files in the local folder via SFTP
-
- Returns:
- List of file paths
- """
- files = []
- # _get_files_recursive is called from _fetch_documents which handles connection
- self._get_files_recursive(self.config.folder_path, files)
- return files
-
- def _get_files_recursive(self, folder_path: str, files: List[str]):
- """
- Recursively get all files in the local folder via SFTP
-
- Args:
- folder_path: Current folder path
- files: List to store found files
- """
- try:
- items = self._sftp_client.listdir_attr(folder_path)
-
- for item in items:
- item_path = os.path.join(folder_path, item.filename)
-
- if item.filename not in ('.', '..'):
- if item.st_mode & 0o040000: # Check if it's a directory
- if self.config.recursive:
- self._get_files_recursive(item_path, files)
- else:
- # Check if file should be ignored
- if not self._should_ignore_file(item_path):
- files.append(item_path)
- except Exception as e:
- logger.error(f"Error listing local folder {folder_path}: {e}")
-
- def _should_ignore_file(self, file_path: str) -> bool:
- """
- Check if file should be ignored based on ignore patterns
-
- Args:
- file_path: File path to check
-
- Returns:
- True if file should be ignored, False otherwise
- """
- if not hasattr(self.config, 'ignore_patterns') or not self.config.ignore_patterns:
- return False
-
- # Get relative path from folder root
- relative_path = os.path.relpath(file_path, self.config.folder_path)
-
- for pattern in self.config.ignore_patterns:
- if self._match_pattern(relative_path, pattern):
- return True
-
- return False
-
- def _match_pattern(self, path: str, pattern: str) -> bool:
- """
- Match a path against a pattern (similar to .gitignore)
-
- Args:
- path: Path to match
- pattern: Pattern to match against
-
- Returns:
- True if path matches pattern, False otherwise
- """
- # Convert glob pattern to regex
- regex_pattern = pattern
- regex_pattern = regex_pattern.replace('.', r'\.')
- regex_pattern = regex_pattern.replace('*', r'.*')
- regex_pattern = regex_pattern.replace('?', r'.')
-
- # Handle directory patterns
- if pattern.endswith('/'):
- regex_pattern = f'^{regex_pattern}.*$'
- else:
- regex_pattern = f'^{regex_pattern}$'
-
- return bool(re.match(regex_pattern, path))
-
- @staticmethod
- def check_data_source_exists(config: BaseDataSourceConfig) -> bool:
- """
- Check if the local folder exists and is accessible via SFTP
-
- Args:
- config: Local folder configuration
-
- Returns:
- True if local folder exists and is accessible, False otherwise
- """
- import paramiko
-
- ssh_client = None
- sftp_client = None
- try:
- ssh_client = paramiko.SSHClient()
- ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
- # Connect to SSH server - use SSH agent if available, otherwise password
- username = config.username or ''
- ssh_client.connect(
- hostname=config.host,
- port=config.port or 22,
- username=username,
- password=config.password,
- timeout=10,
- allow_agent=True, # 允许使用SSH代理
- look_for_keys=False # 禁用查找本地密钥文件
- )
-
- sftp_client = ssh_client.open_sftp()
- sftp_client.stat(config.folder_path)
- return True
- except Exception as e:
- logger.error(f"Error checking local folder via SFTP: {e}")
- return False
- finally:
- if sftp_client:
- sftp_client.close()
- if ssh_client:
- ssh_client.close()
diff --git a/sync/mysql_sync.py b/sync/mysql_sync.py
index c916ddb..e2135a5 100644
--- a/sync/mysql_sync.py
+++ b/sync/mysql_sync.py
@@ -51,46 +51,9 @@ class MySQLSync(BaseSync):
database=self.db_config.database
)
- def fetch_all_documents(self) -> List[Dict[str, Any]]:
+ def fetch_all_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
"""
- Fetch all documents from the MySQL database
-
- Returns:
- List of documents
- """
- return self._fetch_documents()
-
- def fetch_new_documents(self, last_sync_time=None, synced_doc_ids=None) -> List[Dict[str, Any]]:
- """
- Fetch new/updated documents from the MySQL database since last sync time
-
- Args:
- last_sync_time: Last synchronization time
- synced_doc_ids: Set of document IDs that have already been synced (ignored for MySQL)
-
- Returns:
- List of new/updated documents
- """
- return self._fetch_documents(last_sync_time)
-
- def get_synced_document_ids(self) -> Set[str]:
- """
- Get IDs of all documents in the MySQL database
-
- Returns:
- Set of document IDs
- """
- cursor = self.connection.cursor(pymysql.cursors.DictCursor)
- try:
- query = f"SELECT {self.db_config.id_column} FROM {self.db_config.table_name}"
- cursor.execute(query)
- return {str(row[self.db_config.id_column]) for row in cursor.fetchall()}
- finally:
- cursor.close()
-
- def _fetch_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
- """
- Internal method to fetch documents from MySQL
+ Fetch documents from the MySQL database
Args:
last_sync_time: Last synchronization time (for incremental sync)
@@ -143,62 +106,123 @@ class MySQLSync(BaseSync):
if file_content:
row[self.db_config.content_column] = file_content
+ # Generate unique document ID
+ record_id = str(row[self.db_config.id_column])
+ doc_id = self.generate_doc_id(record_id)
+ row['id'] = doc_id
+
+ # Check if document has already been synced
+ if self.vector_store_manager and self.vector_store_manager.document_exists(doc_id):
+ # If document is already synced, check if it's been modified since last sync
+ if last_sync_time and self.db_config.updated_at_column and row.get(self.db_config.updated_at_column):
+ # Skip if not modified since last sync
+ if row[self.db_config.updated_at_column] <= last_sync_time:
+ continue
+ elif last_sync_time:
+ # No updated_at column, skip since we can't determine if it's been modified
+ continue
+
documents.append(row)
return documents
finally:
cursor.close()
- def _load_file_content(self, file_path: str) -> Optional[str]:
+ def fetch_new_documents(self, last_sync_time=None) -> List[Dict[str, Any]]:
"""
- Load file content using FileParser class
+ Fetch new/updated documents from the MySQL database since last sync time
Args:
- file_path: Path to the file
+ last_sync_time: Last synchronization time
Returns:
- File content if successfully loaded, None otherwise
+ List of new/updated documents
"""
- try:
- if not self.db_config.file_source_type:
- logger.debug(f"No file source type configured, skipping file loading for {file_path}")
- return None
-
- # 获取文件扩展名
- import os
- ext = os.path.splitext(file_path)[1].lower()
-
- # 检查文件是否被支持
- if ext not in FileParser.SUPPORTED_EXTENSIONS:
- logger.debug(f"Unsupported file type {ext} for {file_path}, skipping...")
- return None
-
- # 准备文件源配置
- source_config = {
- 'file_system_base_path': self.db_config.file_system_base_path,
- 'scp_host': self.db_config.scp_host,
- 'scp_port': self.db_config.scp_port,
- 'scp_username': self.db_config.scp_username,
- 'scp_password': self.db_config.scp_password,
- 'scp_key_path': self.db_config.scp_key_path
- }
-
- # 使用 BaseSync 中的通用方法获取文件字节
- file_bytes = self._get_file_bytes(file_path, self.db_config.file_source_type, source_config)
-
- if file_bytes:
- # 对于 SCP 文件源，传递 SCP 主机地址
- host = None
- if self.db_config.file_source_type == 'scp' and self.db_config.scp_host:
- host = self.db_config.scp_host
- # 使用 BaseSync 中的通用方法解析文件内容
- return self._parse_file_content(file_bytes, file_path, host=host)
-
- except Exception as e:
- logger.error(f"Error loading file content for {file_path}: {e}")
- return None
+ # Call fetch_all_documents which now handles document existence checks
+ return self.fetch_all_documents(last_sync_time)
+
+ def get_synced_document_ids(self) -> Set[str]:
+ """
+ Get IDs of all documents in the MySQL database
- return None
+ Returns:
+ Set of document IDs
+ """
+ cursor = self.connection.cursor(pymysql.cursors.DictCursor)
+ try:
+ query = f"SELECT {self.db_config.id_column} FROM {self.db_config.table_name}"
+ cursor.execute(query)
+ return {str(row[self.db_config.id_column]) for row in cursor.fetchall()}
+ finally:
+ cursor.close()
+
+ def generate_doc_id(self, record_id: str) -> str:
+ """
+ Generate a unique document ID for MySQL records
+
+ Args:
+ record_id: ID of the record in the database
+
+ Returns:
+ Unique document ID based on database, table, and record ID
+ """
+ # 为 MySQL 记录生成唯一的文档 ID
+ return f"{self.db_config.database}_{self.db_config.table_name}_{record_id}"
+
+ def doc_to_llamaindex_doc(self, doc: Dict) -> 'Document':
+ """
+ Convert MySQL document to LlamaIndex Document
+
+ Args:
+ doc: MySQL document dictionary
+
+ Returns:
+ LlamaIndex Document object
+ """
+ from llama_index.core import Document
+
+ # 处理多个 content 列（支持合并多个列的内容）
+ if self.db_config:
+ # 使用配置的多个 content 列
+ content_columns = self.db_config.content_columns
+ content_separator = self.db_config.content_separator
+ else:
+ # 向后兼容：使用单个 content_column
+ content_columns = ["content"] # Default to "content" column
+ content_separator = "\n"
+
+ # 合并所有 content 列的内容
+ content_parts = []
+ for col in content_columns:
+ col_value = doc.get(col, "")
+ if col_value:
+ content_parts.append(str(col_value))
+
+ # 用指定的分隔符连接多个列的内容
+ content = content_separator.join(content_parts) if content_parts else ""
+
+ title = doc.get('title', "")
+ doc_id = doc.get('id', "")
+
+ # Build metadata
+ metadata = {
+ "doc_id": doc_id,
+ "source": "mysql",
+ "database": self.db_config.database,
+ "table": self.db_config.table_name
+ }
+
+ if title:
+ metadata["title"] = title
+
+ # Create Document
+ return Document(
+ text=content,
+ id_=doc_id,
+ metadata=metadata
+ )
+
+
@staticmethod
def check_data_source_exists(config: DatabaseConfig) -> bool:
diff --git a/sync/remote_folder_sync.py b/sync/remote_folder_sync.py
deleted file mode 100644
index 396a047..0000000
--- a/sync/remote_folder_sync.py
+++ /dev/null
@@ -1,286 +0,0 @@
-"""Remote folder synchronization implementation"""
-import os
-import re
-from typing import List, Dict, Any, Set
-from datetime import datetime
-from pathlib import Path
-from loguru import logger
-from config import BaseDataSourceConfig
-from sync.base_sync import BaseSync
-from rag.file_parser import FileParser
-
-
-class RemoteFolderSync(BaseSync):
- """Handle synchronization between remote folder (via SCP) and ChromaDB"""
-
- def __init__(self, config: BaseDataSourceConfig):
- """
- Initialize remote folder sync with configuration
-
- Args:
- config: Remote folder configuration
- """
- super().__init__(config)
- self.file_parser = FileParser()
- self._ssh_client = None
- self._sftp_client = None
-
- def fetch_all_documents(self) -> List[Dict[str, Any]]:
- """
- Fetch all documents from the remote folder
-
- Returns:
- List of documents
- """
- return self._fetch_documents()
-
- def fetch_new_documents(self, last_sync_time=None, synced_doc_ids=None) -> List[Dict[str, Any]]:
- """
- Fetch new/updated documents from the remote folder since last sync time
-
- Args:
- last_sync_time: Last synchronization time
- synced_doc_ids: Set of document IDs that have already been synced
-
- Returns:
- List of new/updated documents
- """
- return self._fetch_documents(last_sync_time, synced_doc_ids)
-
- def get_synced_document_ids(self) -> Set[str]:
- """
- Get IDs of all files in the remote folder
-
- Returns:
- Set of file paths (as document IDs)
- """
- files = self._get_all_files()
- return {str(file) for file in files}
-
- def _fetch_documents(self, last_sync_time=None, synced_doc_ids=None) -> List[Dict[str, Any]]:
- """
- Internal method to fetch documents from remote folder via SFTP
-
- Args:
- last_sync_time: Last synchronization time (for incremental sync)
- synced_doc_ids: Set of document IDs that have already been synced
-
- Returns:
- List of documents
- """
- documents = []
- self._connect()
- try:
- files = self._get_all_files()
-
- for file_path in files:
- # Check if document has already been synced
- if synced_doc_ids and file_path in synced_doc_ids:
- # If file is already synced, check if it's been modified since last sync
- if last_sync_time:
- file_stat = self._sftp_client.stat(file_path)
- file_mtime = datetime.fromtimestamp(file_stat.st_mtime)
- # Skip if not modified since last sync
- if file_mtime <= last_sync_time:
- continue
- # If file is not synced yet, always include it regardless of modification time
- # This handles the case where files were added to the folder after last_sync_time but have older mtimes
-
- # Parse file content
- try:
- # 使用 BaseSync 中的通用方法获取文件字节
- # 对于远程文件，我们需要先通过 SFTP 获取文件内容
- with self._sftp_client.open(file_path, 'rb') as f:
- file_bytes = f.read()
-
- # 检查文件扩展名是否在支持的列表中
- if Path(file_path).suffix.lower() not in FileParser.SUPPORTED_EXTENSIONS:
- logger.debug(f"Skipping unsupported file: {file_path}")
- continue
-
- # 使用 BaseSync 中的通用方法解析文件内容，传递远程主机地址
- content = self._parse_file_content(file_bytes, file_path, host=self.config.host) if file_bytes else f"[无法读取文件：{Path(file_path).name}]"
-
- document = {
- 'id': str(file_path),
- 'title': Path(file_path).name,
- 'content': content,
- 'file_path': str(file_path),
- 'update_time': datetime.fromtimestamp(self._sftp_client.stat(file_path).st_mtime)
- }
- documents.append(document)
- except Exception as e:
- logger.error(f"Error processing file {file_path}: {e}")
- finally:
- self._disconnect()
-
- return documents
-
- def _get_all_files(self) -> List[str]:
- """
- Get all files in the remote folder
-
- Returns:
- List of file paths
- """
- files = []
- self._get_files_recursive(self.config.folder_path, files)
- return files
-
- def _get_files_recursive(self, folder_path: str, files: List[str]):
- """
- Recursively get all files in the remote folder
-
- Args:
- folder_path: Current folder path
- files: List to store found files
- """
- try:
- items = self._sftp_client.listdir_attr(folder_path)
-
- for item in items:
- item_path = os.path.join(folder_path, item.filename)
-
- if item.filename not in ('.', '..'):
- if item.st_mode & 0o040000: # Check if it's a directory
- if self.config.recursive:
- self._get_files_recursive(item_path, files)
- else:
- # Check if file should be ignored
- if not self._should_ignore_file(item_path):
- files.append(item_path)
- except Exception as e:
- logger.error(f"Error listing remote folder {folder_path}: {e}")
-
- def _should_ignore_file(self, file_path: str) -> bool:
- """
- Check if file should be ignored based on ignore patterns
-
- Args:
- file_path: File path to check
-
- Returns:
- True if file should be ignored, False otherwise
- """
- if not hasattr(self.config, 'ignore_patterns') or not self.config.ignore_patterns:
- return False
-
- # Get relative path from folder root
- relative_path = os.path.relpath(file_path, self.config.folder_path)
-
- for pattern in self.config.ignore_patterns:
- if self._match_pattern(relative_path, pattern):
- return True
-
- return False
-
- def _match_pattern(self, path: str, pattern: str) -> bool:
- """
- Match a path against a pattern (similar to .gitignore)
-
- Args:
- path: Path to match
- pattern: Pattern to match against
-
- Returns:
- True if path matches pattern, False otherwise
- """
- # Convert glob pattern to regex
- regex_pattern = pattern
- regex_pattern = regex_pattern.replace('.', r'\.')
- regex_pattern = regex_pattern.replace('*', r'.*')
- regex_pattern = regex_pattern.replace('?', r'.')
-
- # Handle directory patterns
- if pattern.endswith('/'):
- regex_pattern = f'^{regex_pattern}.*$'
- else:
- regex_pattern = f'^{regex_pattern}$'
-
- return bool(re.match(regex_pattern, path))
-
- def _connect(self):
- """
- Connect to the remote server via SSH/SFTP
- """
- import paramiko
-
- self._ssh_client = paramiko.SSHClient()
- self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
- # Connect to SSH server
- # 确保username不为None，否则paramiko会报错
- username = self.config.username or ''
-
- # 使用密码认证或SSH代理认证
- self._ssh_client.connect(
- hostname=self.config.host,
- port=self.config.port or 22,
- username=username,
- password=self.config.password,
- timeout=10,
- allow_agent=True, # 允许使用SSH代理
- look_for_keys=False # 禁用查找本地密钥文件
- )
-
- # Create SFTP client
- self._sftp_client = self._ssh_client.open_sftp()
-
- def _disconnect(self):
- """
- Disconnect from the remote server
- """
- if self._sftp_client:
- self._sftp_client.close()
- self._sftp_client = None
-
- if self._ssh_client:
- self._ssh_client.close()
- self._ssh_client = None
-
- @staticmethod
- def check_data_source_exists(config: BaseDataSourceConfig) -> bool:
- """
- Check if the remote folder exists and is accessible
-
- Args:
- config: Remote folder configuration
-
- Returns:
- True if remote folder exists and is accessible, False otherwise
- """
- import paramiko
-
- ssh_client = None
- sftp_client = None
-
- try:
- ssh_client = paramiko.SSHClient()
- ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
- # Connect to SSH server - use SSH agent if available, otherwise password
- username = config.username or ''
- ssh_client.connect(
- hostname=config.host,
- port=config.port or 22,
- username=username,
- password=config.password,
- timeout=10,
- allow_agent=True, # 允许使用SSH代理
- look_for_keys=False # 禁用查找本地密钥文件
- )
-
- # Create SFTP client and check folder exists
- sftp_client = ssh_client.open_sftp()
- sftp_client.stat(config.folder_path)
-
- return True
- except Exception as e:
- logger.error(f"Error checking remote folder: {e}")
- return False
- finally:
- if sftp_client:
- sftp_client.close()
- if ssh_client:
- ssh_client.close()
-
diff --git a/sync_service.py b/sync_service.py
index efacb65..5cb55ed 100644
--- a/sync_service.py
+++ b/sync_service.py
@@ -3,11 +3,12 @@ Background service for syncing data from various sources to ChromaDB
"""
import asyncio
from datetime import datetime
-from typing import Set, List, Dict, Any
+from typing import Dict
from loguru import logger
from config import settings, BaseDataSourceConfig
from sync.base_sync import BaseSync, get_sync_class
-from rag import VectorStoreManager, DocumentProcessor
+from rag import VectorStoreManager
+from db_utils import get_data_source_update_at, update_data_source_update_at
class SyncService:
@@ -19,11 +20,17 @@ class SyncService:
# Check if the data source exists before proceeding
logger.info(f"Checking if data source exists: {self.source_name}")
- all_exist, missing_sources = BaseSync.check_data_sources_exist([self.source_config])
- if not all_exist:
+ try:
+ # 直接调用具体同步类的 check_data_source_exists 方法，以便捕获详细的错误信息
+ sync_class = get_sync_class(self.source_config.type)
+ sync_class.check_data_source_exists(self.source_config)
+
+ logger.info(f"✓ Data source {self.source_name} exists")
+ except Exception as e:
error_msg = (
f"Error: Data source {self.source_name} does not exist or cannot be accessed\n"
+ f"Details: {str(e)}\n"
f"Please check:\n"
f" 1. Data source exists and is accessible\n"
f" 2. Connection details (host, port, credentials) are correct\n"
@@ -33,10 +40,7 @@ class SyncService:
logger.error(error_msg)
raise RuntimeError(error_msg)
- logger.info(f"✓ Data source {self.source_name} exists")
-
self.vector_store_manager = VectorStoreManager()
- self.document_processor = DocumentProcessor()
self._running = False
self._sync_in_progress = False # Flag to prevent concurrent syncs
self._auto_sync_task = None # Reference to auto sync task to prevent multiple instances
@@ -47,11 +51,19 @@ class SyncService:
sync_class = get_sync_class(self.source_config.type)
# Create a new syncer instance for this data source
- self.syncer = sync_class(self.source_config)
+ self.syncer = sync_class(self.source_config, vector_store_manager=self.vector_store_manager)
# Initialize sync tracking data
self.last_sync_time = None
- self.synced_doc_ids = set()
+
+ # Read last sync time from data_sources table if available
+ try:
+ update_at = get_data_source_update_at(self.source_name)
+ if update_at:
+ self.last_sync_time = update_at
+ logger.info(f"Initialized last_sync_time from data_sources: {self.last_sync_time}")
+ except Exception as e:
+ logger.warning(f"Error reading last_sync_time from data_sources: {e}")
logger.info(f"Initialized {self.source_config.type} syncer for {self.source_name}")
except Exception as e:
@@ -82,10 +94,7 @@ class SyncService:
def sync_work():
"""Synchronous work that runs in thread pool"""
- # Get existing document IDs from ChromaDB to avoid re-processing
- existing_doc_ids = set()
- if not force:
- existing_doc_ids = self.vector_store_manager.get_existing_doc_ids()
+ # Will check document existence individually using document_exists() method
all_chunked_docs = [] # 存储所有分块后的文档
total_docs = 0
@@ -103,6 +112,7 @@ class SyncService:
documents = self.syncer.fetch_all_documents()
if not documents:
+ self.last_sync_time = datetime.now()
return all_chunked_docs, total_docs, skipped_docs_count
# Database-specific processing for content columns
@@ -136,16 +146,13 @@ class SyncService:
doc_id = str(doc.get(id_column, ""))
# 生成唯一文档标识符（数据库标识名称_表名_文档ID）{db_source}_{table_name}_{id}
unique_doc_id = f"{self.source_config.name}_{self.source_config.table_name}_{doc_id}"
- if unique_doc_id not in existing_doc_ids:
+ if not self.vector_store_manager.document_exists(unique_doc_id):
new_documents.append(doc)
else:
db_skipped_count += 1
skipped_docs_count += 1
if not new_documents:
- self.synced_doc_ids = {
- str(doc.get(id_column)) for doc in documents
- }
self.last_sync_time = datetime.now()
return all_chunked_docs, total_docs, skipped_docs_count
documents = new_documents
@@ -153,17 +160,13 @@ class SyncService:
self.vector_store_manager.delete_documents_by_source(source_identifier)
# Process and chunk documents
- processed_docs = self.document_processor.process_documents(documents, self.source_config)
- chunked_docs = self.document_processor.chunk_documents(processed_docs)
+ processed_docs = self.syncer.process_documents(documents)
+ chunked_docs = self.syncer.chunk_documents(processed_docs)
all_chunked_docs.extend(chunked_docs)
- # Update synced document IDs
- id_column = self.source_config.id_column
- self.synced_doc_ids = {
- str(doc.get(id_column)) for doc in documents
- }
- else:
- # Non-database sources (local/remote folder)
+ # Update last sync time
+ self.last_sync_time = datetime.now()
+ elif self.source_config.type == "folder":
if not force:
new_documents = []
for doc in documents:
@@ -172,31 +175,35 @@ class SyncService:
logger.info(f"Sync interrupted during document filtering: {self.source_name}")
return all_chunked_docs, total_docs, skipped_docs_count # 返回结果，不退出程序
- doc_id = doc.get('id', str(doc.get('file_path', '')))
- if doc_id not in existing_doc_ids:
+ doc_id = doc.get('id')
+ if not self.vector_store_manager.document_exists(doc_id):
new_documents.append(doc)
else:
skipped_docs_count += 1
if not new_documents:
- self.synced_doc_ids = {
- doc.get('id', str(doc.get('file_path', ''))) for doc in documents
- }
self.last_sync_time = datetime.now()
return all_chunked_docs, total_docs, skipped_docs_count
documents = new_documents
# Process and chunk documents
- processed_docs = self.document_processor.process_documents(documents, self.source_config)
- chunked_docs = self.document_processor.chunk_documents(processed_docs)
+ processed_docs = self.syncer.process_documents(documents)
+ chunked_docs = self.syncer.chunk_documents(processed_docs)
all_chunked_docs.extend(chunked_docs)
- # Update synced document IDs
- self.synced_doc_ids = {
- doc.get('id', str(doc.get('file_path', ''))) for doc in documents
- }
+ # Update last sync time
+ self.last_sync_time = datetime.now()
+ else:
+ raise Exception(f"不支持的数据类型")
self.last_sync_time = datetime.now()
+
+ # Update update_at in data_sources table
+ try:
+ update_data_source_update_at(self.source_name, self.last_sync_time)
+ except Exception as e:
+ logger.warning(f"Error updating update_at in data_sources: {e}")
+
total_docs += len(documents)
except Exception as e:
@@ -269,29 +276,27 @@ class SyncService:
logger.info(f"Incremental sync interrupted: {self.source_name}")
return all_chunked_docs, total_docs # 返回结果，不退出程序
- # Fetch new documents, passing synced_doc_ids to determine what needs syncing
- new_documents = self.syncer.fetch_new_documents(self.last_sync_time, self.synced_doc_ids)
+ # Fetch new documents
+ new_documents = self.syncer.fetch_new_documents(self.last_sync_time)
if not new_documents:
logger.debug(f"No new documents in data source: {self.source_name}")
return all_chunked_docs, total_docs
# Process and chunk documents
- processed_docs = self.document_processor.process_documents(new_documents, self.source_config)
- chunked_docs = self.document_processor.chunk_documents(processed_docs)
+ processed_docs = self.syncer.process_documents(new_documents)
+ chunked_docs = self.syncer.chunk_documents(processed_docs)
all_chunked_docs.extend(chunked_docs)
- # Update synced document IDs for this data source
- if self.source_config.type == 'database':
- id_column = self.source_config.id_column
- new_doc_ids = {str(doc.get(id_column)) for doc in new_documents}
- self.synced_doc_ids.update(new_doc_ids)
- else:
- # Use file path or id as document ID for folder sources
- new_doc_ids = {doc.get('id', str(doc.get('file_path', ''))) for doc in new_documents}
- self.synced_doc_ids.update(new_doc_ids)
-
+ # Update last sync time
self.last_sync_time = datetime.now()
+
+ # Update update_at in data_sources table
+ try:
+ update_data_source_update_at(self.source_name, self.last_sync_time)
+ except Exception as e:
+ logger.warning(f"Error updating update_at in data_sources: {e}")
+
total_docs += len(new_documents)
logger.info(f"Incremental sync: {len(chunked_docs)} chunks from {len(new_documents)} documents in {self.source_config.type}: {self.source_name}")
@@ -388,23 +393,9 @@ class SyncService:
logger.info(f"Auto sync service with recovery stopped for {self.source_name}")
self._running = False
- async def start_auto_sync(self, skip_initial_sync: bool = False):
- """
- Start automatic periodic sync in background for this data source.
- This method runs continuously until stop_auto_sync() is called.
- For production use, prefer start_auto_sync_with_recovery() which includes error recovery.
-
- Args:
- skip_initial_sync: If True, skip the initial sync_all() call.
- Use this when initial sync is already done elsewhere.
- """
- logger.info(f"Starting auto sync for data source: {self.source_name}")
- await self._run_auto_sync_loop(skip_initial_sync)
-
async def _run_auto_sync_loop(self, skip_initial_sync: bool = False):
"""
Internal method that runs the auto sync loop for this data source.
- This is separated so it can be called by both start_auto_sync and start_auto_sync_with_recovery.
Args:
skip_initial_sync: If True, skip the initial sync_all() call.