feat(rag): 优化 vllm 作为 LLM 提供商支持
- 在 RAGEngine 中实现 vllm 提供商的支持 - 添加 OpenAILike 集成用于 vllm 兼容性 - 更新 docker-compose.yml 中的默认 LLM 配置为 vllm - 添加 llama-index-llms-openai-like 依赖 - 优化同步模块中的文件路径处理逻辑 - 修复 SSH 连接和 SFTP 文件操作的路径兼容性问题
This commit is contained in:
Bennie61
parent
1cb82f4216
commit
8401d90efb
|
|
@ -1,5 +1,5 @@
|
|||
# Docker Compose 配置说明
|
||||
#
|
||||
#
|
||||
# 网络模式: 使用 host 网络模式，容器直接使用宿主机的网络
|
||||
# - 优点: 可以直接访问宿主机上的服务（MySQL、Ollama 等），无需额外网络配置
|
||||
# - 注意: 使用 host 网络模式时，不能使用 ports 映射，容器直接使用宿主机的端口
|
||||
|
|
@ -28,6 +28,7 @@ services:
|
|||
# ChromaDB 容器默认将数据存储在 /data 目录
|
||||
- ./chroma_db_data:/data
|
||||
# 将主机端口映射为可配置（从 .env 读取 CHROMA_SERVER_PORT，默认 8002）
|
||||
# 注意以下 8000端口是ChromaDB默认端口，不用更改
|
||||
ports:
|
||||
- "${CHROMA_SERVER_PORT:-8002}:8000"
|
||||
environment:
|
||||
|
|
@ -73,37 +74,44 @@ services:
|
|||
- API_PORT=${API_PORT:-8001}
|
||||
- API_TITLE=${API_TITLE:-RAG API}
|
||||
- API_VERSION=${API_VERSION:-1.0.0}
|
||||
|
||||
|
||||
# ChromaDB 配置
|
||||
- CHROMA_SERVER_HOST=${CHROMA_SERVER_HOST:-localhost}
|
||||
- CHROMA_SERVER_PORT=${CHROMA_SERVER_PORT:-8002}
|
||||
- CHROMA_COLLECTION_NAME=${CHROMA_COLLECTION_NAME:-rag_collection}
|
||||
|
||||
|
||||
# MySQL 配置
|
||||
- MYSQL_HOST=${MYSQL_HOST:-localhost}
|
||||
- MYSQL_PORT=${MYSQL_PORT:-3306}
|
||||
- MYSQL_USER=${MYSQL_USER:-root}
|
||||
- MYSQL_PASSWORD=${MYSQL_PASSWORD:-}
|
||||
- MYSQL_DATABASE=${MYSQL_DATABASE:-forgeplus}
|
||||
|
||||
|
||||
# 多数据库配置（可选）
|
||||
- MYSQL_DATABASES_CONFIG=${MYSQL_DATABASES_CONFIG:-}
|
||||
|
||||
|
||||
# Ollama 配置
|
||||
- OLLAMA_BASE_URL=${OLLAMA_BASE_URL:-http://localhost:11434}
|
||||
- OLLAMA_MODEL=${OLLAMA_MODEL:-qwen3:1.7b}
|
||||
- OLLAMA_EMBEDDING_MODEL=${OLLAMA_EMBEDDING_MODEL:-qwen3-embedding:0.6b}
|
||||
|
||||
|
||||
# LLM 配置
|
||||
# LLM provider: ollama, openai 等
|
||||
- LLM_PROVIDER=${LLM_PROVIDER:-vllm}
|
||||
- LLM_BASE_URL=${LLM_BASE_URL:-http://172.20.32.236:8000/v1}
|
||||
- LLM_MODEL=${LLM_MODEL:-MiniMax-M2.7}
|
||||
- LLM_API_KEY=${LLM_API_KEY:-none} # 如使用openai等需要API Key的服务
|
||||
|
||||
# RAG 配置
|
||||
- EMBEDDING_DIMENSION=${EMBEDDING_DIMENSION:-768}
|
||||
- CHUNK_SIZE=${CHUNK_SIZE:-1024}
|
||||
- CHUNK_OVERLAP=${CHUNK_OVERLAP:-200}
|
||||
- TOP_K=${TOP_K:-5}
|
||||
|
||||
|
||||
# 同步配置
|
||||
- SYNC_INTERVAL=${SYNC_INTERVAL:-300}
|
||||
- AUTO_SYNC=${AUTO_SYNC:-true}
|
||||
|
||||
|
||||
# NLTK 配置
|
||||
- NLTK_DATA=${NLTK_DATA:-./nltk_data/}
|
||||
# 注意：使用 host 网络模式时，depends_on 可能无法正常工作
|
||||
|
|
|
|||
|
|
@ -81,18 +81,18 @@ QA_PROMPT_NO_HISTORY = PromptTemplate(QA_PROMPT_STR_NO_HISTORY)
|
|||
|
||||
class RAGEngine:
|
||||
"""Main RAG engine for query processing"""
|
||||
|
||||
|
||||
@staticmethod
|
||||
def check_llm_connection(provider: str, base_url: str, model: str, api_key: Optional[str] = None) -> Tuple[bool, str]:
|
||||
"""
|
||||
Check if LLM server is accessible and connection can be established
|
||||
|
||||
|
||||
Args:
|
||||
provider: LLM provider (ollama, vllm, openai)
|
||||
base_url: Base URL for the LLM service
|
||||
model: Model name
|
||||
api_key: Optional API key for authentication
|
||||
|
||||
|
||||
Returns:
|
||||
Tuple of (is_connected: bool, error_message: str)
|
||||
If connected, error_message will be empty string
|
||||
|
|
@ -101,7 +101,7 @@ class RAGEngine:
|
|||
headers = {}
|
||||
if api_key:
|
||||
headers["Authorization"] = f"Bearer {api_key}"
|
||||
|
||||
|
||||
if provider == "ollama":
|
||||
logger.info(f"Checking Ollama connection to {base_url}...")
|
||||
with httpx.Client(timeout=10.0) as client:
|
||||
|
|
@ -134,22 +134,22 @@ class RAGEngine:
|
|||
else:
|
||||
logger.warning(f"Unknown LLM provider: {provider}, skipping connection check")
|
||||
return True, ""
|
||||
|
||||
|
||||
except httpx.ConnectError as e:
|
||||
error_message = f"Cannot connect to {provider} server at {base_url}. " \
|
||||
f"Please check if server is running and accessible."
|
||||
f"Please check if server is running and accessible."
|
||||
logger.error(f"✗ {provider} connection failed: {error_message}")
|
||||
return False, error_message
|
||||
except httpx.TimeoutException:
|
||||
error_message = f"Connection to {provider} server at {base_url} timed out. " \
|
||||
f"Please check if server is running and accessible."
|
||||
f"Please check if server is running and accessible."
|
||||
logger.error(f"✗ {provider} connection failed: {error_message}")
|
||||
return False, error_message
|
||||
except Exception as e:
|
||||
error_message = f"Unexpected error while checking {provider} connection: {str(e)}"
|
||||
logger.error(f"✗ {provider} connection check failed: {error_message}")
|
||||
return False, error_message
|
||||
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
vector_store_manager: VectorStoreManager,
|
||||
|
|
@ -163,13 +163,13 @@ class RAGEngine:
|
|||
self._system_prompt = system_prompt
|
||||
self._temperature = temperature
|
||||
self._request_timeout = request_timeout
|
||||
|
||||
|
||||
llm_config = settings.get_llm_config()
|
||||
provider = llm_config["provider"]
|
||||
base_url = llm_config["base_url"]
|
||||
model = llm_config["model"]
|
||||
api_key = llm_config["api_key"]
|
||||
|
||||
|
||||
is_connected, error_message = self.check_llm_connection(provider, base_url, model, api_key)
|
||||
if not is_connected:
|
||||
error_msg = (
|
||||
|
|
@ -184,7 +184,7 @@ class RAGEngine:
|
|||
)
|
||||
logger.error(error_msg)
|
||||
raise RuntimeError(error_msg)
|
||||
|
||||
|
||||
if provider == "ollama":
|
||||
self.llm = Ollama(
|
||||
model=model,
|
||||
|
|
@ -200,6 +200,17 @@ class RAGEngine:
|
|||
temperature=self._temperature,
|
||||
timeout=self._request_timeout,
|
||||
)
|
||||
elif provider == "vllm":
|
||||
from llama_index.llms.openai_like import OpenAILike
|
||||
self.llm = OpenAILike(
|
||||
model=model,
|
||||
api_base=base_url,
|
||||
api_key=api_key or "none",
|
||||
is_chat_model=True,
|
||||
context_window=128000,
|
||||
temperature=self._temperature,
|
||||
timeout=self._request_timeout,
|
||||
)
|
||||
else:
|
||||
self.llm = Ollama(
|
||||
model=model,
|
||||
|
|
@ -207,7 +218,7 @@ class RAGEngine:
|
|||
temperature=self._temperature,
|
||||
request_timeout=self._request_timeout,
|
||||
)
|
||||
|
||||
|
||||
self._llm_provider = provider
|
||||
|
||||
def extract_text_from_chunk(self, chunk) -> Optional[str]:
|
||||
|
|
@ -229,12 +240,12 @@ class RAGEngine:
|
|||
async def query_stream(self, query: str, history: str, top_k: Optional[int] = None) -> AsyncIterator[str]:
|
||||
"""
|
||||
Query the RAG system and stream the response
|
||||
|
||||
|
||||
Args:
|
||||
query: User query string
|
||||
history: Chat history string
|
||||
top_k: Number of documents to retrieve (optional)
|
||||
|
||||
|
||||
Yields:
|
||||
Response text chunks
|
||||
"""
|
||||
|
|
@ -243,7 +254,7 @@ class RAGEngine:
|
|||
retriever = self.vector_store_manager.get_retriever(top_k=top_k or settings.TOP_K)
|
||||
# query index
|
||||
retrieved_nodes = await retriever.aretrieve(query)
|
||||
|
||||
|
||||
# 2. 构建上下文
|
||||
context_parts = []
|
||||
for i, node in enumerate(retrieved_nodes[:top_k or settings.TOP_K], 1): # 限制数量
|
||||
|
|
@ -253,47 +264,47 @@ class RAGEngine:
|
|||
if len(text) > 400:
|
||||
text = text[:400] + "..."
|
||||
context_parts.append(f"【参考信息{i}】{text}")
|
||||
|
||||
|
||||
context_str = "\n\n".join(context_parts) if context_parts else "未找到相关参考信息"
|
||||
|
||||
|
||||
if history is not None:
|
||||
qa_prompt = QA_PROMPT_HISTORY
|
||||
filled_prompt = qa_prompt.format(history=history, context_str=context_str, query_str=query)
|
||||
else:
|
||||
qa_prompt = QA_PROMPT_NO_HISTORY
|
||||
filled_prompt = qa_prompt.format(context_str=context_str, query_str=query)
|
||||
|
||||
|
||||
stream_response = await self.llm.astream_complete(
|
||||
prompt=filled_prompt
|
||||
)
|
||||
|
||||
)
|
||||
|
||||
full_response = ""
|
||||
think_filter = OptimizedDeltaThinkFilter()
|
||||
|
||||
|
||||
async for chunk in stream_response:
|
||||
# 提取文本内容
|
||||
# text_chunk = self.extract_text_from_chunk(chunk)
|
||||
delta, full_text, has_output = think_filter.process_delta_robust(chunk)
|
||||
|
||||
|
||||
if delta is not None:
|
||||
full_response += delta
|
||||
yield delta.encode('utf-8')
|
||||
await asyncio.sleep(0.001) # slight delay to yield control
|
||||
|
||||
|
||||
logger.info(f"响应完成，长度: {len(full_response)}字符")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in RAG query: {e}")
|
||||
|
||||
|
||||
async def query(self, query: str, history: str, top_k: Optional[int] = None) -> str:
|
||||
"""
|
||||
Query the RAG system and return complete response
|
||||
|
||||
|
||||
Args:
|
||||
query: User query string
|
||||
history: Chat history string
|
||||
top_k: Number of documents to retrieve (optional)
|
||||
|
||||
|
||||
Returns:
|
||||
Complete response string
|
||||
"""
|
||||
|
|
@ -302,7 +313,7 @@ class RAGEngine:
|
|||
retriever = self.vector_store_manager.get_retriever(top_k=top_k or settings.TOP_K)
|
||||
# query index
|
||||
retrieved_nodes = await retriever.aretrieve(query)
|
||||
|
||||
|
||||
# 2. 构建上下文
|
||||
context_parts = []
|
||||
for i, node in enumerate(retrieved_nodes[:top_k or settings.TOP_K], 1): # 限制数量
|
||||
|
|
@ -312,16 +323,16 @@ class RAGEngine:
|
|||
if len(text) > 400:
|
||||
text = text[:400] + "..."
|
||||
context_parts.append(f"【参考信息{i}】{text}")
|
||||
|
||||
|
||||
context_str = "\n\n".join(context_parts) if context_parts else "未找到相关参考信息"
|
||||
|
||||
|
||||
if history is not None:
|
||||
qa_prompt = QA_PROMPT_HISTORY
|
||||
filled_prompt = qa_prompt.format(history=history, context_str=context_str, query_str=query)
|
||||
else:
|
||||
qa_prompt = QA_PROMPT_NO_HISTORY
|
||||
filled_prompt = qa_prompt.format(context_str=context_str, query_str=query)
|
||||
|
||||
|
||||
response = await self.llm.acomplete(
|
||||
prompt=filled_prompt
|
||||
)
|
||||
|
|
|
|||
|
|
@ -42,4 +42,6 @@ markdown2>=2.5.4 # Markdown parsing
|
|||
# Remote file access
|
||||
paramiko>=3.4.0 # For SSH/SCP functionality
|
||||
|
||||
requests~=2.32.5
|
||||
requests~=2.32.5
|
||||
|
||||
llama-index-llms-openai-like>=0.2.0
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
"""Folder synchronization implementation for local and remote folders"""
|
||||
import os
|
||||
import posixpath
|
||||
import re
|
||||
from typing import List, Dict, Any, Set
|
||||
from datetime import datetime
|
||||
|
|
@ -14,11 +15,11 @@ import paramiko
|
|||
|
||||
class SSHClient:
|
||||
"""SSH client for testing connections to remote servers"""
|
||||
|
||||
|
||||
def __init__(self, host, port=22, username=None, password=None):
|
||||
"""
|
||||
Initialize SSH client
|
||||
|
||||
|
||||
Args:
|
||||
host: Host address
|
||||
port: Port number
|
||||
|
|
@ -30,11 +31,11 @@ class SSHClient:
|
|||
self.username = username
|
||||
self.password = password
|
||||
self._client = None
|
||||
|
||||
|
||||
def test_connection(self):
|
||||
"""
|
||||
Test SSH connection
|
||||
|
||||
|
||||
Returns:
|
||||
bool: True if connection is successful, False otherwise
|
||||
"""
|
||||
|
|
@ -42,7 +43,7 @@ class SSHClient:
|
|||
# Create SSH client
|
||||
self._client = paramiko.SSHClient()
|
||||
self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||||
|
||||
|
||||
# Connect with timeout
|
||||
self._client.connect(
|
||||
hostname=self.host,
|
||||
|
|
@ -53,7 +54,7 @@ class SSHClient:
|
|||
allow_agent=True,
|
||||
look_for_keys=False
|
||||
)
|
||||
|
||||
|
||||
# Connection successful
|
||||
return True
|
||||
finally:
|
||||
|
|
@ -64,11 +65,11 @@ class SSHClient:
|
|||
|
||||
class FolderSync(BaseSync):
|
||||
"""Handle synchronization between folder (local or remote) and ChromaDB"""
|
||||
|
||||
|
||||
def __init__(self, config: BaseDataSourceConfig, vector_store_manager=None):
|
||||
"""
|
||||
Initialize folder sync with configuration
|
||||
|
||||
|
||||
Args:
|
||||
config: Folder configuration
|
||||
"""
|
||||
|
|
@ -77,14 +78,14 @@ class FolderSync(BaseSync):
|
|||
self._ssh_client = None
|
||||
self._sftp_client = None
|
||||
self.vector_store_manager = vector_store_manager
|
||||
|
||||
|
||||
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
|
||||
"""
|
||||
|
|
@ -92,19 +93,19 @@ class FolderSync(BaseSync):
|
|||
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 + "_chunk_0"):
|
||||
|
|
@ -115,11 +116,11 @@ class FolderSync(BaseSync):
|
|||
# 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:
|
||||
|
|
@ -136,7 +137,7 @@ class FolderSync(BaseSync):
|
|||
content = f"[无法读取文件：{Path(file_path).name}]"
|
||||
else:
|
||||
content = f"[无法读取文件：{Path(file_path).name}]"
|
||||
|
||||
|
||||
# Build document
|
||||
document = {
|
||||
'id': doc_id,
|
||||
|
|
@ -152,22 +153,22 @@ class FolderSync(BaseSync):
|
|||
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
|
||||
|
|
@ -181,68 +182,68 @@ class FolderSync(BaseSync):
|
|||
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,
|
||||
|
|
@ -253,11 +254,11 @@ class FolderSync(BaseSync):
|
|||
'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:
|
||||
|
|
@ -266,7 +267,7 @@ class FolderSync(BaseSync):
|
|||
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
|
||||
|
|
@ -274,18 +275,18 @@ class FolderSync(BaseSync):
|
|||
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
|
||||
"""
|
||||
|
|
@ -296,21 +297,22 @@ class FolderSync(BaseSync):
|
|||
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)
|
||||
|
||||
# old:relative_path = os.path.relpath(file_path, self.config.folder_path)
|
||||
item_path = posixpath.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:
|
||||
|
|
@ -321,37 +323,38 @@ class FolderSync(BaseSync):
|
|||
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)
|
||||
|
||||
# old: relative_path = os.path.relpath(file_path, self.config.folder_path)
|
||||
relative_path = posixpath.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
|
||||
"""
|
||||
|
|
@ -360,45 +363,45 @@ class FolderSync(BaseSync):
|
|||
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(
|
||||
|
|
@ -416,14 +419,14 @@ class FolderSync(BaseSync):
|
|||
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}")
|
||||
|
|
|
|||
Loading…
Reference in New Issue