682 lines
35 KiB
Python
682 lines
35 KiB
Python
"""
|
||
Background service for syncing data from various sources to ChromaDB
|
||
"""
|
||
import asyncio
|
||
from datetime import datetime
|
||
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
|
||
from db_utils import get_data_source_update_at, update_data_source_update_at
|
||
|
||
|
||
class SyncService:
|
||
"""Service for synchronizing data from various sources (database, local_folder, remote_folder) to ChromaDB"""
|
||
|
||
def __init__(self, source_config: BaseDataSourceConfig):
|
||
self.source_config = source_config
|
||
self.source_name = source_config.name
|
||
|
||
self.vector_store_manager = VectorStoreManager()
|
||
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
|
||
self.syncer = None # Will be initialized in sync_all
|
||
|
||
# Initialize sync tracking data
|
||
self.last_sync_time = None
|
||
|
||
# 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 SyncService for {self.source_name}")
|
||
|
||
async def sync_all(self, force: bool = False, is_manual: bool = False):
|
||
"""
|
||
Sync all documents from this data source to ChromaDB
|
||
|
||
Args:
|
||
force: If True, re-process all documents even if they exist (default: False)
|
||
is_manual: If True, this sync was triggered manually, ignore _running flag (default: False)
|
||
"""
|
||
# Prevent concurrent syncs
|
||
if self._sync_in_progress:
|
||
logger.warning("Sync already in progress, skipping this request")
|
||
return
|
||
|
||
self._sync_in_progress = True
|
||
sync_start_time = datetime.now()
|
||
try:
|
||
logger.info(f"[同步] 开始全量同步: {self.source_name}")
|
||
|
||
# Initialize syncer if not already initialized
|
||
if not self.syncer:
|
||
logger.info(f"Initializing syncer for {self.source_name}...")
|
||
# Check if the data source exists before proceeding
|
||
try:
|
||
# 直接调用具体同步类的 check_data_source_exists 方法，以便捕获详细的错误信息
|
||
if self.source_config.type == 'database':
|
||
sync_class = get_sync_class(self.source_config.type, getattr(self.source_config, 'db_type', 'mysql'))
|
||
else:
|
||
sync_class = get_sync_class(self.source_config.type)
|
||
sync_class.check_data_source_exists(self.source_config)
|
||
|
||
logger.info(f"✓ Data source {self.source_name} exists")
|
||
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"
|
||
f" 3. Network connectivity is available\n"
|
||
f" 4. Firewall rules allow connections"
|
||
)
|
||
logger.error(error_msg)
|
||
raise RuntimeError(error_msg)
|
||
|
||
# Initialize syncer for this data source
|
||
try:
|
||
# Get the appropriate sync class based on data source type
|
||
if self.source_config.type == 'database':
|
||
sync_class = get_sync_class(self.source_config.type, getattr(self.source_config, 'db_type', 'mysql'))
|
||
else:
|
||
sync_class = get_sync_class(self.source_config.type)
|
||
|
||
# Create a new syncer instance for this data source
|
||
self.syncer = sync_class(self.source_config, self.vector_store_manager)
|
||
|
||
logger.info(f"Initialized {self.source_config.type} syncer for {self.source_name}")
|
||
except Exception as e:
|
||
logger.error(f"Failed to initialize sync for {self.source_name}: {e}")
|
||
raise
|
||
|
||
# Run all synchronous operations in thread pool to avoid blocking event loop
|
||
import asyncio
|
||
loop = asyncio.get_event_loop()
|
||
|
||
def sync_work():
|
||
"""Synchronous work that runs in thread pool"""
|
||
# Will check document existence individually using document_exists() method
|
||
|
||
all_chunked_docs = [] # 存储所有分块后的文档
|
||
total_docs = 0
|
||
skipped_docs_count = 0
|
||
|
||
# Sync from this data source
|
||
try:
|
||
# 检查服务运行状态：
|
||
# - 自动同步：必须检查_running标志，确保服务没有被停止
|
||
# - 手动同步：忽略_running标志，允许在服务停止后执行
|
||
if not is_manual and not self._running:
|
||
logger.info(f"Sync interrupted: {self.source_name}")
|
||
return all_chunked_docs, total_docs, skipped_docs_count # 返回结果，不退出程序
|
||
|
||
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
|
||
if self.source_config.type == 'database':
|
||
# 从配置中获取内容列
|
||
content_columns = self.source_config.content_columns
|
||
|
||
# 获取每个数据源对应 ChromaDB metadata 中的 “content_column”
|
||
source_identifier = f"{self.source_config.database}_{self.source_config.table_name}"
|
||
db_content_column_str = self.vector_store_manager.get_specific_db_source_metadata(source_identifier)
|
||
if db_content_column_str:
|
||
existing_db_content_columns = [col.strip() for col in db_content_column_str.split(",")]
|
||
# 获取 existing_doc_content_columns 与 content_columns 差异项
|
||
diff_columns = set(content_columns) - set(existing_db_content_columns)
|
||
else: # 说明 ChromaDB 中无该数据源的 metadata，需全量处理，无需过滤
|
||
diff_columns = set(content_columns)
|
||
|
||
# Filter out documents that already exist (if not forcing)
|
||
db_skipped_count = 0
|
||
if not force:
|
||
id_column = self.source_config.id_column
|
||
# 若 diff_columns为空，说明无新增列，则需过滤ChromaDB中已存在的doc
|
||
if not diff_columns:
|
||
new_documents = []
|
||
for doc in documents:
|
||
# 检查服务运行状态：仅在非手动同步时检查
|
||
if not is_manual and not self._running:
|
||
logger.info(f"Sync interrupted during document filtering: {self.source_name}")
|
||
return all_chunked_docs, total_docs, skipped_docs_count # 返回结果，不退出程序
|
||
|
||
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 not self.vector_store_manager.document_exists(unique_doc_id + "_chunk_0"):
|
||
new_documents.append(doc)
|
||
else:
|
||
db_skipped_count += 1
|
||
skipped_docs_count += 1
|
||
|
||
if not new_documents:
|
||
self.last_sync_time = datetime.now()
|
||
return all_chunked_docs, total_docs, skipped_docs_count
|
||
documents = new_documents
|
||
elif db_content_column_str:
|
||
self.vector_store_manager.delete_documents_by_source(source_identifier)
|
||
|
||
# Process and chunk documents
|
||
processed_docs = self.syncer.process_documents(documents)
|
||
chunked_docs = self.syncer.chunk_documents(processed_docs)
|
||
all_chunked_docs.extend(chunked_docs)
|
||
|
||
# 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:
|
||
# 检查服务运行状态：仅在非手动同步时检查
|
||
if not is_manual and not self._running:
|
||
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')
|
||
if not self.vector_store_manager.document_exists(doc_id):
|
||
new_documents.append(doc)
|
||
else:
|
||
skipped_docs_count += 1
|
||
|
||
if not new_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.syncer.process_documents(documents)
|
||
chunked_docs = self.syncer.chunk_documents(processed_docs)
|
||
all_chunked_docs.extend(chunked_docs)
|
||
|
||
# Update last sync time
|
||
self.last_sync_time = datetime.now()
|
||
elif self.source_config.type == "git":
|
||
if not force:
|
||
new_documents = []
|
||
for doc in documents:
|
||
# 检查服务运行状态：仅在非手动同步时检查
|
||
if not is_manual and not self._running:
|
||
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')
|
||
if not self.vector_store_manager.document_exists(doc_id):
|
||
new_documents.append(doc)
|
||
else:
|
||
skipped_docs_count += 1
|
||
|
||
if not new_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.syncer.process_documents(documents)
|
||
chunked_docs = self.syncer.chunk_documents(processed_docs)
|
||
all_chunked_docs.extend(chunked_docs)
|
||
|
||
# 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:
|
||
logger.error(f"[同步] 数据源 {self.source_name} 同步失败: {e}")
|
||
raise
|
||
|
||
return all_chunked_docs, total_docs, skipped_docs_count
|
||
|
||
# Run synchronous work in thread pool
|
||
all_chunked_docs, total_docs, skipped_docs_count = await loop.run_in_executor(None, sync_work)
|
||
|
||
# Add all documents to vector store (run in thread pool to avoid blocking event loop)
|
||
if all_chunked_docs:
|
||
await loop.run_in_executor(
|
||
None,
|
||
self.vector_store_manager.add_documents,
|
||
all_chunked_docs,
|
||
not force # skip_existing true
|
||
)
|
||
|
||
sync_duration = (datetime.now() - sync_start_time).total_seconds()
|
||
logger.info(
|
||
f"[同步] 完成: {self.source_name}, "
|
||
f"{total_docs} 个新文档, {len(all_chunked_docs)} 个分块"
|
||
+ (f", 跳过 {skipped_docs_count} 个已存在" if skipped_docs_count > 0 else "")
|
||
+ f", 耗时 {sync_duration:.1f}秒"
|
||
)
|
||
else:
|
||
sync_duration = (datetime.now() - sync_start_time).total_seconds()
|
||
if skipped_docs_count > 0:
|
||
logger.info(f"[同步] 完成: {self.source_name}, 所有 {skipped_docs_count} 个文档已存在, 耗时 {sync_duration:.1f}秒")
|
||
else:
|
||
logger.warning(f"[同步] 没有文档需要同步: {self.source_name}, 耗时 {sync_duration:.1f}秒")
|
||
except Exception as e:
|
||
sync_duration = (datetime.now() - sync_start_time).total_seconds()
|
||
logger.error(f"[同步进度] ✗ {self.source_name} 同步过程中出错 (耗时: {sync_duration:.1f} 秒): {e}")
|
||
raise
|
||
finally:
|
||
self._sync_in_progress = False
|
||
|
||
async def sync_incremental(self, is_manual: bool = False):
|
||
"""
|
||
Sync only new/updated documents from this data source to ChromaDB
|
||
|
||
Args:
|
||
is_manual: If True, this sync was triggered manually, ignore _running flag (default: False)
|
||
"""
|
||
# Prevent concurrent syncs
|
||
if self._sync_in_progress:
|
||
logger.warning("Sync already in progress, skipping incremental sync")
|
||
return
|
||
|
||
self._sync_in_progress = True
|
||
try:
|
||
logger.info(f"Starting incremental sync from data source: {self.source_name}")
|
||
|
||
# Initialize syncer if not already initialized
|
||
if not self.syncer:
|
||
logger.info(f"Initializing syncer for {self.source_name}...")
|
||
# Check if the data source exists before proceeding
|
||
try:
|
||
# 直接调用具体同步类的 check_data_source_exists 方法，以便捕获详细的错误信息
|
||
if self.source_config.type == 'database':
|
||
sync_class = get_sync_class(self.source_config.type, getattr(self.source_config, 'db_type', 'mysql'))
|
||
else:
|
||
sync_class = get_sync_class(self.source_config.type)
|
||
sync_class.check_data_source_exists(self.source_config)
|
||
|
||
logger.info(f"✓ Data source {self.source_name} exists")
|
||
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"
|
||
f" 3. Network connectivity is available\n"
|
||
f" 4. Firewall rules allow connections"
|
||
)
|
||
logger.error(error_msg)
|
||
raise RuntimeError(error_msg)
|
||
|
||
# Initialize syncer for this data source
|
||
try:
|
||
# Get the appropriate sync class based on data source type
|
||
if self.source_config.type == 'database':
|
||
sync_class = get_sync_class(self.source_config.type, getattr(self.source_config, 'db_type', 'mysql'))
|
||
else:
|
||
sync_class = get_sync_class(self.source_config.type)
|
||
|
||
# Create a new syncer instance for this data source
|
||
self.syncer = sync_class(self.source_config, self.vector_store_manager)
|
||
|
||
logger.info(f"Initialized {self.source_config.type} syncer for {self.source_name}")
|
||
except Exception as e:
|
||
logger.error(f"Failed to initialize sync for {self.source_name}: {e}")
|
||
raise
|
||
|
||
# Run all synchronous operations in thread pool
|
||
import asyncio
|
||
loop = asyncio.get_event_loop()
|
||
|
||
def sync_work():
|
||
"""Synchronous work that runs in thread pool"""
|
||
all_chunked_docs = []
|
||
total_docs = 0
|
||
|
||
# Sync from this data source
|
||
try:
|
||
# 检查服务运行状态：仅在非手动同步时检查
|
||
if not is_manual and not self._running:
|
||
logger.info(f"Incremental sync interrupted: {self.source_name}")
|
||
return all_chunked_docs, total_docs # 返回结果，不退出程序
|
||
|
||
# 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.syncer.process_documents(new_documents)
|
||
chunked_docs = self.syncer.chunk_documents(processed_docs)
|
||
all_chunked_docs.extend(chunked_docs)
|
||
|
||
# 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}")
|
||
except Exception as e:
|
||
logger.error(f"Error during incremental sync for data source {self.source_name}: {e}")
|
||
raise
|
||
|
||
return all_chunked_docs, total_docs
|
||
|
||
# Run synchronous work in thread pool
|
||
all_chunked_docs, total_docs = await loop.run_in_executor(None, sync_work)
|
||
|
||
# Add all new documents to vector store
|
||
if all_chunked_docs:
|
||
# Run in thread pool to avoid blocking event loop
|
||
await loop.run_in_executor(
|
||
None,
|
||
self.vector_store_manager.add_documents,
|
||
all_chunked_docs,
|
||
False # skip_existing，不跳过已存在的文档，默认是更新了内容
|
||
)
|
||
logger.info(f"Incremental sync completed: {len(all_chunked_docs)} chunks from {total_docs} documents in {self.source_name}")
|
||
else:
|
||
logger.info(f"No new documents to sync in {self.source_name}")
|
||
except Exception as e:
|
||
logger.error(f"Error during incremental sync for {self.source_name}: {e}")
|
||
raise
|
||
finally:
|
||
self._sync_in_progress = False
|
||
|
||
async def start_auto_sync_with_recovery(self, skip_initial_sync: bool = False):
|
||
"""
|
||
Start automatic periodic sync in background with error recovery for this data source.
|
||
If the sync service stops due to an error, it will automatically restart.
|
||
This method runs continuously until stop_auto_sync() is called.
|
||
|
||
Args:
|
||
skip_initial_sync: If True, skip the initial sync_all() call.
|
||
Use this when initial sync is already done elsewhere.
|
||
"""
|
||
if not settings.AUTO_SYNC:
|
||
logger.info(f"Auto sync is disabled for data source: {self.source_name}")
|
||
return
|
||
|
||
# Prevent multiple auto sync service instances
|
||
if self._running:
|
||
logger.warning(f"Auto sync service is already running for data source: {self.source_name}, skipping duplicate start request")
|
||
return
|
||
|
||
# Check if there's an existing auto sync task still running
|
||
if self._auto_sync_task is not None and not self._auto_sync_task.done():
|
||
logger.warning(f"Auto sync task is still running for data source: {self.source_name}, skipping duplicate start request")
|
||
return
|
||
|
||
max_restart_attempts = 10 # Maximum number of restart attempts
|
||
restart_delay = 60 # Wait 60 seconds before restarting after an error
|
||
restart_count = 0
|
||
|
||
self._running = True
|
||
logger.info(f"Auto sync service with error recovery started for {self.source_name} (interval: {settings.SYNC_INTERVAL}s)")
|
||
|
||
while self._running and restart_count < max_restart_attempts:
|
||
try:
|
||
# Store the task reference to prevent multiple instances
|
||
self._auto_sync_task = asyncio.create_task(
|
||
self._run_auto_sync_loop(skip_initial_sync)
|
||
)
|
||
await self._auto_sync_task
|
||
# If we reach here, the loop exited normally (not due to error)
|
||
logger.info(f"Auto sync loop exited normally for {self.source_name}")
|
||
break
|
||
except Exception as e:
|
||
restart_count += 1
|
||
logger.error(
|
||
f"Auto sync service stopped due to error for {self.source_name} (restart attempt {restart_count}/{max_restart_attempts}): {e}",
|
||
exc_info=True
|
||
)
|
||
|
||
if restart_count >= max_restart_attempts:
|
||
logger.error(f"Auto sync service failed after {max_restart_attempts} restart attempts for {self.source_name}. Stopping auto sync.")
|
||
self._running = False
|
||
break
|
||
|
||
if not self._running:
|
||
logger.info(f"Auto sync service stop requested for {self.source_name}, not restarting")
|
||
break
|
||
|
||
logger.info(f"Waiting {restart_delay}s before restarting auto sync service for {self.source_name}...")
|
||
await asyncio.sleep(restart_delay)
|
||
logger.info(f"Restarting auto sync service for {self.source_name} (attempt {restart_count + 1}/{max_restart_attempts})...")
|
||
# Reset skip_initial_sync after first attempt (only skip on first start)
|
||
skip_initial_sync = False
|
||
|
||
logger.info(f"Auto sync service with recovery stopped for {self.source_name}")
|
||
self._running = False
|
||
|
||
async def _run_auto_sync_loop(self, skip_initial_sync: bool = False):
|
||
"""
|
||
Internal method that runs the auto sync loop for this data source.
|
||
|
||
Args:
|
||
skip_initial_sync: If True, skip the initial sync_all() call.
|
||
"""
|
||
if not settings.AUTO_SYNC:
|
||
logger.info(f"Auto sync is disabled for {self.source_name}")
|
||
return
|
||
|
||
logger.info(f"Auto sync loop started for {self.source_name} (interval: {settings.SYNC_INTERVAL}s)")
|
||
|
||
# Initial sync (only if not skipped)
|
||
if not skip_initial_sync:
|
||
logger.info(f"Performing initial sync for {self.source_name} in auto sync service...")
|
||
await self.sync_all()
|
||
else:
|
||
logger.info(f"Skipping initial sync for {self.source_name} in auto sync service (already done elsewhere)")
|
||
|
||
# Periodic incremental sync
|
||
sync_count = 0
|
||
last_sync_start_time = None
|
||
while self._running:
|
||
try:
|
||
logger.info(f"Auto sync waiting {settings.SYNC_INTERVAL}s before next sync for {self.source_name} (count: {sync_count})...")
|
||
await asyncio.sleep(settings.SYNC_INTERVAL)
|
||
if not self._running:
|
||
logger.info(f"Auto sync stopped for {self.source_name}, exiting loop")
|
||
break
|
||
|
||
# Check if another sync is in progress (e.g., initial sync or previous incremental sync still running)
|
||
# Wait for it to complete before starting incremental sync (no timeout - wait indefinitely)
|
||
wait_interval = 10 # Check every 10 seconds
|
||
waited_time = 0
|
||
while self._sync_in_progress:
|
||
logger.info(f"Another sync is in progress for {self.source_name}, waiting... (waited {waited_time}s, will wait until completion)")
|
||
await asyncio.sleep(wait_interval)
|
||
waited_time += wait_interval
|
||
|
||
# Log warning if waiting for a very long time (for monitoring purposes)
|
||
if waited_time % 300 == 0: # Every 5 minutes
|
||
logger.info(f"Still waiting for sync to complete for {self.source_name}... (waited {waited_time}s / {waited_time // 60} minutes)")
|
||
|
||
# If last sync started more than 2 hours ago and is still running, log warning
|
||
if last_sync_start_time is not None:
|
||
time_since_last_sync = (datetime.now() - last_sync_start_time).total_seconds()
|
||
if time_since_last_sync > 7200: # 2 hours
|
||
logger.warning(f"Last incremental sync for {self.source_name} has been running for {time_since_last_sync / 3600:.1f} hours, this might indicate a slow sync. Continuing to wait...")
|
||
|
||
if waited_time > 0:
|
||
logger.info(f"Previous sync completed for {self.source_name}, waited {waited_time}s / {waited_time // 60} minutes")
|
||
|
||
if not self._running:
|
||
logger.info(f"Auto sync stopped during wait for {self.source_name}, exiting loop")
|
||
break
|
||
|
||
# Record sync start time for monitoring
|
||
last_sync_start_time = datetime.now()
|
||
sync_count += 1
|
||
logger.info(f"Running periodic incremental sync #{sync_count} for {self.source_name}...")
|
||
|
||
# Execute incremental sync - it will check _sync_in_progress internally
|
||
try:
|
||
await self.sync_incremental()
|
||
logger.info(f"Incremental sync #{sync_count} completed successfully for {self.source_name}")
|
||
except Exception as sync_error:
|
||
logger.error(f"Incremental sync #{sync_count} failed for {self.source_name}: {sync_error}", exc_info=True)
|
||
# Reset sync in progress flag if it was set (in case of unexpected error)
|
||
if self._sync_in_progress:
|
||
logger.warning(f"Resetting _sync_in_progress flag for {self.source_name} due to error in incremental sync")
|
||
self._sync_in_progress = False
|
||
# Continue to next cycle even if this sync failed
|
||
logger.info(f"Continuing to next sync cycle for {self.source_name} despite error...")
|
||
continue
|
||
|
||
except asyncio.CancelledError:
|
||
logger.info(f"Auto sync task was cancelled for {self.source_name}")
|
||
break
|
||
except Exception as e:
|
||
logger.error(f"Unexpected error in auto sync loop for {self.source_name}: {e}", exc_info=True)
|
||
# Reset sync in progress flag if it was set (in case of unexpected error)
|
||
if self._sync_in_progress:
|
||
logger.warning(f"Resetting _sync_in_progress flag for {self.source_name} due to unexpected error in auto sync loop")
|
||
self._sync_in_progress = False
|
||
# Continue running even if there's an error
|
||
logger.info(f"Continuing auto sync loop for {self.source_name} despite error...")
|
||
continue
|
||
|
||
logger.info(f"Auto sync loop stopped for {self.source_name} (total syncs: {sync_count})")
|
||
|
||
def stop_auto_sync(self):
|
||
"""Stop automatic sync for this data source"""
|
||
if not self._running:
|
||
logger.info(f"Auto sync service is not running for {self.source_name}")
|
||
return
|
||
|
||
self._running = False
|
||
logger.info(f"Auto sync service stop requested for {self.source_name}")
|
||
|
||
# Cancel the auto sync task if it exists and is running
|
||
if hasattr(self, '_auto_sync_task') and self._auto_sync_task is not None and not self._auto_sync_task.done():
|
||
try:
|
||
self._auto_sync_task.cancel()
|
||
logger.info(f"Cancelled auto sync task for {self.source_name}")
|
||
except Exception as e:
|
||
logger.error(f"Error cancelling auto sync task for {self.source_name}: {e}")
|
||
|
||
# 不再等待当前同步完成，直接设置标志位并返回
|
||
# 同步操作内部会定期检查self._running标志
|
||
logger.info(f"Auto sync service stopped for {self.source_name}, current sync will be interrupted if running")
|
||
|
||
def close(self):
|
||
"""Close connections for this data source"""
|
||
self.stop_auto_sync()
|
||
if hasattr(self.syncer, 'close'):
|
||
self.syncer.close()
|
||
|
||
|
||
class SyncServiceManager:
|
||
"""Manager for multiple SyncService instances, one per data source"""
|
||
|
||
def __init__(self):
|
||
self.data_sources = settings.get_data_sources()
|
||
self.sync_services: Dict[str, SyncService] = {}
|
||
|
||
# Initialize sync services for each data source
|
||
for source_config in self.data_sources:
|
||
try:
|
||
sync_service = SyncService(source_config)
|
||
self.sync_services[source_config.name] = sync_service
|
||
logger.info(f"Created SyncService for {source_config.name}")
|
||
except Exception as e:
|
||
logger.error(f"Failed to create SyncService for {source_config.name}: {e}")
|
||
|
||
async def start_all_sync_services(self):
|
||
"""Start auto sync with recovery for all data sources"""
|
||
tasks = []
|
||
for sync_service in self.sync_services.values():
|
||
# Start each sync service in background
|
||
task = asyncio.create_task(sync_service.start_auto_sync_with_recovery())
|
||
tasks.append(task)
|
||
|
||
# Wait for all tasks to complete (they should run indefinitely until stopped)
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
def stop_all_sync_services(self):
|
||
"""Stop all sync services"""
|
||
for sync_service in self.sync_services.values():
|
||
sync_service.stop_auto_sync()
|
||
logger.info("All sync services stopped")
|
||
|
||
def close_all(self):
|
||
"""Close all sync services"""
|
||
for sync_service in self.sync_services.values():
|
||
sync_service.close()
|
||
logger.info("All sync services closed")
|
||
|
||
def get_sync_service(self, source_name: str) -> SyncService:
|
||
"""Get a SyncService instance for a specific data source"""
|
||
return self.sync_services.get(source_name)
|
||
|
||
def create_or_update_sync_service(self, source_config: BaseDataSourceConfig):
|
||
"""Create or update a SyncService for a data source"""
|
||
source_name = source_config.name
|
||
|
||
# Stop and remove existing sync service if it exists
|
||
if source_name in self.sync_services:
|
||
self.sync_services[source_name].close()
|
||
del self.sync_services[source_name]
|
||
logger.info(f"Removed existing SyncService for {source_name}")
|
||
|
||
# Create new sync service
|
||
try:
|
||
sync_service = SyncService(source_config)
|
||
self.sync_services[source_name] = sync_service
|
||
logger.info(f"Created new SyncService for {source_name}")
|
||
return sync_service
|
||
except Exception as e:
|
||
logger.error(f"Failed to create SyncService for {source_name}: {e}")
|
||
raise
|
||
|
||
def remove_sync_service(self, source_name: str):
|
||
"""Remove a SyncService for a data source"""
|
||
if source_name in self.sync_services:
|
||
self.sync_services[source_name].close()
|
||
del self.sync_services[source_name]
|
||
logger.info(f"Removed SyncService for {source_name}")
|
||
|
||
|
||
def main():
|
||
"""Main function to run the sync service"""
|
||
import asyncio
|
||
|
||
# Create sync service manager and start all sync services
|
||
sync_manager = SyncServiceManager()
|
||
try:
|
||
asyncio.run(sync_manager.start_all_sync_services())
|
||
except KeyboardInterrupt:
|
||
sync_manager.stop_all_sync_services()
|
||
sync_manager.close_all()
|
||
print("Sync service manager stopped by user")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|