| """数据库抽象层 - 支持文件系统和远程SQL数据库，以及双向同步""" |
| import os |
| import json |
| import asyncio |
| import hashlib |
| from abc import ABC, abstractmethod |
| from typing import List, Dict, Any, Optional, Callable |
| from pathlib import Path |
| import logging |
| import time |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class DatabaseInterface(ABC): |
| """数据库接口抽象类""" |
|
|
| @abstractmethod |
| async def save_accounts(self, accounts: List[Dict[str, Any]]) -> bool: |
| """保存账号配置""" |
| pass |
|
|
| @abstractmethod |
| async def load_accounts(self) -> List[Dict[str, Any]]: |
| """加载账号配置""" |
| pass |
|
|
| @abstractmethod |
| async def save_config(self, config: Dict[str, Any]) -> bool: |
| """保存完整配置""" |
| pass |
|
|
| @abstractmethod |
| async def load_config(self) -> Dict[str, Any]: |
| """加载完整配置""" |
| pass |
|
|
| @abstractmethod |
| async def save_admin_config(self, admin_config: Dict[str, Any]) -> bool: |
| """保存管理员配置""" |
| pass |
|
|
| @abstractmethod |
| async def load_admin_config(self) -> Dict[str, Any]: |
| """加载管理员配置""" |
| pass |
|
|
| @abstractmethod |
| async def initialize(self) -> bool: |
| """初始化数据库""" |
| pass |
|
|
| @abstractmethod |
| async def get_config_hash(self) -> str: |
| """获取配置哈希值用于变更检测""" |
| pass |
|
|
|
|
| class FileSystemDatabase(DatabaseInterface): |
| """文件系统数据库实现（原有逻辑）""" |
|
|
| def __init__(self, data_dir: Path): |
| self.data_dir = data_dir |
| self.config_file = data_dir / "config.json" |
| self.admin_config_file = data_dir / "admin.json" |
|
|
| def _ensure_dir(self): |
| """确保目录存在""" |
| self.data_dir.mkdir(parents=True, exist_ok=True) |
|
|
| async def initialize(self) -> bool: |
| """初始化文件系统""" |
| try: |
| self._ensure_dir() |
| return True |
| except Exception as e: |
| logger.error(f"文件系统初始化失败: {e}") |
| return False |
|
|
| async def save_accounts(self, accounts: List[Dict[str, Any]]) -> bool: |
| """保存账号配置""" |
| try: |
| self._ensure_dir() |
| config = await self.load_config() |
| config["accounts"] = accounts |
| return await self.save_config(config) |
| except Exception as e: |
| logger.error(f"保存账号配置失败: {e}") |
| return False |
|
|
| async def load_accounts(self) -> List[Dict[str, Any]]: |
| """加载账号配置""" |
| config = await self.load_config() |
| return config.get("accounts", []) |
|
|
| async def save_config(self, config: Dict[str, Any]) -> bool: |
| """保存完整配置""" |
| try: |
| self._ensure_dir() |
| with open(self.config_file, "w", encoding="utf-8") as f: |
| json.dump(config, f, indent=2, ensure_ascii=False) |
| return True |
| except Exception as e: |
| logger.error(f"保存配置失败: {e}") |
| return False |
|
|
| async def load_config(self) -> Dict[str, Any]: |
| """加载完整配置""" |
| try: |
| if self.config_file.exists(): |
| with open(self.config_file, "r", encoding="utf-8") as f: |
| return json.load(f) |
| except Exception as e: |
| logger.error(f"加载配置失败: {e}") |
| return {} |
|
|
| async def save_admin_config(self, admin_config: Dict[str, Any]) -> bool: |
| """保存管理员配置""" |
| try: |
| self._ensure_dir() |
| with open(self.admin_config_file, "w", encoding="utf-8") as f: |
| json.dump(admin_config, f, indent=2, ensure_ascii=False) |
| return True |
| except Exception as e: |
| logger.error(f"保存管理员配置失败: {e}") |
| return False |
|
|
| async def load_admin_config(self) -> Dict[str, Any]: |
| """加载管理员配置""" |
| try: |
| if self.admin_config_file.exists(): |
| with open(self.admin_config_file, "r", encoding="utf-8") as f: |
| return json.load(f) |
| except Exception as e: |
| logger.error(f"加载管理员配置失败: {e}") |
| return {} |
|
|
| async def get_config_hash(self) -> str: |
| """获取配置文件的哈希值""" |
| try: |
| if self.config_file.exists(): |
| content = self.config_file.read_bytes() |
| return hashlib.md5(content).hexdigest() |
| except Exception as e: |
| logger.error(f"获取配置哈希失败: {e}") |
| return "" |
|
|
|
|
| class SQLDatabase(DatabaseInterface): |
| """SQL数据库实现""" |
|
|
| def __init__(self, database_url: str): |
| self.database_url = database_url |
| self.pool = None |
| self._db_type = self._detect_db_type(database_url) |
|
|
| def _detect_db_type(self, url: str) -> str: |
| """检测数据库类型""" |
| if url.startswith("postgresql://") or url.startswith("postgres://"): |
| return "postgresql" |
| elif url.startswith("mysql://") or url.startswith("mysql+"): |
| return "mysql" |
| elif url.startswith("sqlite://"): |
| return "sqlite" |
| else: |
| return "unknown" |
|
|
| async def initialize(self) -> bool: |
| """初始化SQL数据库连接和表结构""" |
| try: |
| if self._db_type == "postgresql": |
| await self._init_postgresql() |
| elif self._db_type == "mysql": |
| await self._init_mysql() |
| elif self._db_type == "sqlite": |
| await self._init_sqlite() |
| else: |
| logger.error(f"不支持的数据库类型: {self._db_type}") |
| return False |
|
|
| await self._create_tables() |
| return True |
| except Exception as e: |
| logger.error(f"SQL数据库初始化失败: {e}") |
| return False |
|
|
| async def _init_postgresql(self): |
| """初始化PostgreSQL连接""" |
| try: |
| import asyncpg |
| self.pool = await asyncpg.create_pool(self.database_url) |
| except ImportError: |
| raise ImportError("请安装 asyncpg: pip install asyncpg") |
|
|
| async def _init_mysql(self): |
| """初始化MySQL连接""" |
| try: |
| import aiomysql |
| |
| from urllib.parse import urlparse |
| parsed = urlparse(self.database_url) |
| self.pool = await aiomysql.create_pool( |
| host=parsed.hostname, |
| port=parsed.port or 3306, |
| user=parsed.username, |
| password=parsed.password, |
| db=parsed.path.lstrip('/'), |
| charset='utf8mb4' |
| ) |
| except ImportError: |
| raise ImportError("请安装 aiomysql: pip install aiomysql") |
|
|
| async def _init_sqlite(self): |
| """初始化SQLite连接""" |
| try: |
| import aiosqlite |
| db_path = self.database_url.replace("sqlite://", "") |
| self.db_path = db_path |
| except ImportError: |
| raise ImportError("请安装 aiosqlite: pip install aiosqlite") |
|
|
| async def _create_tables(self): |
| """创建数据表""" |
| if self._db_type == "postgresql": |
| await self._create_tables_postgresql() |
| elif self._db_type == "mysql": |
| await self._create_tables_mysql() |
| elif self._db_type == "sqlite": |
| await self._create_tables_sqlite() |
|
|
| async def _create_tables_postgresql(self): |
| """创建PostgreSQL表""" |
| async with self.pool.acquire() as conn: |
| await conn.execute(""" |
| CREATE TABLE IF NOT EXISTS kiro_config ( |
| id SERIAL PRIMARY KEY, |
| key VARCHAR(255) UNIQUE NOT NULL, |
| value JSONB NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ) |
| """) |
|
|
| await conn.execute(""" |
| CREATE TABLE IF NOT EXISTS kiro_admin ( |
| id SERIAL PRIMARY KEY, |
| key VARCHAR(255) UNIQUE NOT NULL, |
| value JSONB NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ) |
| """) |
|
|
| async def _create_tables_mysql(self): |
| """创建MySQL表""" |
| async with self.pool.acquire() as conn: |
| async with conn.cursor() as cursor: |
| await cursor.execute(""" |
| CREATE TABLE IF NOT EXISTS kiro_config ( |
| id INT AUTO_INCREMENT PRIMARY KEY, |
| `key` VARCHAR(255) UNIQUE NOT NULL, |
| value JSON NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP |
| ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 |
| """) |
|
|
| await cursor.execute(""" |
| CREATE TABLE IF NOT EXISTS kiro_admin ( |
| id INT AUTO_INCREMENT PRIMARY KEY, |
| `key` VARCHAR(255) UNIQUE NOT NULL, |
| value JSON NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP |
| ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 |
| """) |
|
|
| async def _create_tables_sqlite(self): |
| """创建SQLite表""" |
| import aiosqlite |
| async with aiosqlite.connect(self.db_path) as db: |
| await db.execute(""" |
| CREATE TABLE IF NOT EXISTS kiro_config ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| key TEXT UNIQUE NOT NULL, |
| value TEXT NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ) |
| """) |
|
|
| await db.execute(""" |
| CREATE TABLE IF NOT EXISTS kiro_admin ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| key TEXT UNIQUE NOT NULL, |
| value TEXT NOT NULL, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ) |
| """) |
| await db.commit() |
|
|
| async def _get_config_value(self, table: str, key: str) -> Optional[Dict[str, Any]]: |
| """从指定表获取配置值""" |
| try: |
| if self._db_type == "postgresql": |
| async with self.pool.acquire() as conn: |
| row = await conn.fetchrow(f"SELECT value FROM {table} WHERE key = $1", key) |
| return row['value'] if row else None |
|
|
| elif self._db_type == "mysql": |
| async with self.pool.acquire() as conn: |
| async with conn.cursor() as cursor: |
| await cursor.execute(f"SELECT value FROM {table} WHERE `key` = %s", (key,)) |
| row = await cursor.fetchone() |
| return json.loads(row[0]) if row else None |
|
|
| elif self._db_type == "sqlite": |
| import aiosqlite |
| async with aiosqlite.connect(self.db_path) as db: |
| cursor = await db.execute(f"SELECT value FROM {table} WHERE key = ?", (key,)) |
| row = await cursor.fetchone() |
| return json.loads(row[0]) if row else None |
|
|
| except Exception as e: |
| logger.error(f"获取配置失败 {table}.{key}: {e}") |
| return None |
|
|
| async def _set_config_value(self, table: str, key: str, value: Dict[str, Any]) -> bool: |
| """设置配置值到指定表""" |
| try: |
| if self._db_type == "postgresql": |
| async with self.pool.acquire() as conn: |
| await conn.execute(f""" |
| INSERT INTO {table} (key, value) VALUES ($1, $2) |
| ON CONFLICT (key) DO UPDATE SET value = $2, updated_at = CURRENT_TIMESTAMP |
| """, key, json.dumps(value)) |
|
|
| elif self._db_type == "mysql": |
| async with self.pool.acquire() as conn: |
| async with conn.cursor() as cursor: |
| await cursor.execute(f""" |
| INSERT INTO {table} (`key`, value) VALUES (%s, %s) |
| ON DUPLICATE KEY UPDATE value = %s, updated_at = CURRENT_TIMESTAMP |
| """, (key, json.dumps(value), json.dumps(value))) |
| await conn.commit() |
|
|
| elif self._db_type == "sqlite": |
| import aiosqlite |
| async with aiosqlite.connect(self.db_path) as db: |
| await db.execute(f""" |
| INSERT OR REPLACE INTO {table} (key, value, updated_at) |
| VALUES (?, ?, CURRENT_TIMESTAMP) |
| """, (key, json.dumps(value))) |
| await db.commit() |
|
|
| return True |
| except Exception as e: |
| logger.error(f"设置配置失败 {table}.{key}: {e}") |
| return False |
|
|
| async def save_accounts(self, accounts: List[Dict[str, Any]]) -> bool: |
| """保存账号配置""" |
| config = await self.load_config() |
| config["accounts"] = accounts |
| return await self.save_config(config) |
|
|
| async def load_accounts(self) -> List[Dict[str, Any]]: |
| """加载账号配置""" |
| config = await self.load_config() |
| return config.get("accounts", []) |
|
|
| async def save_config(self, config: Dict[str, Any]) -> bool: |
| """保存完整配置""" |
| return await self._set_config_value("kiro_config", "main", config) |
|
|
| async def load_config(self) -> Dict[str, Any]: |
| """加载完整配置""" |
| config = await self._get_config_value("kiro_config", "main") |
| return config or {} |
|
|
| async def save_admin_config(self, admin_config: Dict[str, Any]) -> bool: |
| """保存管理员配置""" |
| return await self._set_config_value("kiro_admin", "main", admin_config) |
|
|
| async def load_admin_config(self) -> Dict[str, Any]: |
| """加载管理员配置""" |
| config = await self._get_config_value("kiro_admin", "main") |
| return config or {} |
|
|
| async def get_config_hash(self) -> str: |
| """获取配置的哈希值""" |
| try: |
| config = await self.load_config() |
| content = json.dumps(config, sort_keys=True) |
| return hashlib.md5(content.encode()).hexdigest() |
| except Exception as e: |
| logger.error(f"获取配置哈希失败: {e}") |
| return "" |
|
|
|
|
| def _get_account_key(account: Dict[str, Any]) -> str: |
| """获取账号唯一标识用于去重""" |
| if account.get("email"): |
| return f"email:{account['email']}" |
| elif account.get("token"): |
| return f"token:{account['token'][:32]}" |
| elif account.get("id"): |
| return f"id:{account['id']}" |
| return f"hash:{hash(json.dumps(account, sort_keys=True))}" |
|
|
|
|
| def _merge_accounts(local_accounts: List[Dict[str, Any]], remote_accounts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: |
| """合并本地和远程账号并去重，远程优先""" |
| seen_keys = set() |
| merged = [] |
| |
| for account in remote_accounts: |
| key = _get_account_key(account) |
| if key not in seen_keys: |
| seen_keys.add(key) |
| merged.append(account) |
| |
| for account in local_accounts: |
| key = _get_account_key(account) |
| if key not in seen_keys: |
| seen_keys.add(key) |
| merged.append(account) |
| logger.info(f"从本地合并账号: {key}") |
| |
| return merged |
|
|
|
|
| |
| def create_database() -> DatabaseInterface: |
| """根据环境变量创建数据库实例""" |
| database_url = os.getenv("DATABASE_URL") |
|
|
| if database_url: |
| candidate = SQLDatabase(database_url) |
| if candidate._db_type != "unknown": |
| logger.info(f"使用远程数据库: {candidate._db_type}") |
| return candidate |
| logger.warning("DATABASE_URL 格式不支持，将回退到文件系统数据库") |
| else: |
| from ..config import DATA_DIR |
| logger.info(f"使用文件系统数据库: {DATA_DIR}") |
| return FileSystemDatabase(DATA_DIR) |
| |
| from ..config import DATA_DIR |
| logger.info(f"使用文件系统数据库: {DATA_DIR}") |
| return FileSystemDatabase(DATA_DIR) |
|
|
|
|
| |
| _db_instance: Optional[DatabaseInterface] = None |
| _merge_completed: bool = False |
|
|
| async def get_database() -> DatabaseInterface: |
| """获取数据库实例（单例模式），首次启动时自动合并本地账号到远程""" |
| global _db_instance, _merge_completed |
| |
| if _db_instance is None: |
| candidate = create_database() |
| ok = await candidate.initialize() |
| if not ok: |
| from ..config import DATA_DIR |
| logger.warning(f"数据库初始化失败，将回退到文件系统数据库: {DATA_DIR}") |
| candidate = FileSystemDatabase(DATA_DIR) |
| await candidate.initialize() |
|
|
| _db_instance = candidate |
|
|
| if isinstance(_db_instance, SQLDatabase) and not _merge_completed: |
| await _auto_merge_local_to_remote() |
| _merge_completed = True |
| |
| return _db_instance |
|
|
|
|
| async def _auto_merge_local_to_remote(): |
| """自动将本地账号合并到远程数据库""" |
| global _db_instance |
| |
| try: |
| from ..config import DATA_DIR |
| local_db = FileSystemDatabase(DATA_DIR) |
| await local_db.initialize() |
| |
| local_accounts = await local_db.load_accounts() |
| if not local_accounts: |
| logger.info("本地无账号，跳过合并") |
| return |
| |
| remote_accounts = await _db_instance.load_accounts() |
| |
| merged = _merge_accounts(local_accounts, remote_accounts) |
| |
| if len(merged) > len(remote_accounts): |
| await _db_instance.save_accounts(merged) |
| new_count = len(merged) - len(remote_accounts) |
| logger.info(f"已将 {new_count} 个本地账号合并到远程数据库，共 {len(merged)} 个账号") |
| else: |
| logger.info(f"所有本地账号已存在于远程，无需合并") |
| |
| except Exception as e: |
| logger.error(f"自动合并账号失败: {e}") |
|
|
|
|
| class SyncManager: |
| """双向数据同步管理器""" |
| |
| def __init__(self, local_db: FileSystemDatabase, remote_db: SQLDatabase): |
| self.local_db = local_db |
| self.remote_db = remote_db |
| self._local_hash: str = "" |
| self._remote_hash: str = "" |
| self._sync_lock = asyncio.Lock() |
| self._running = False |
| self._sync_task: Optional[asyncio.Task] = None |
| self._sync_interval = int(os.getenv("SYNC_INTERVAL", "30")) |
| self._callbacks: List[Callable] = [] |
| |
| def register_callback(self, callback: Callable): |
| """注册同步完成回调""" |
| self._callbacks.append(callback) |
| |
| async def _notify_callbacks(self): |
| """通知所有回调""" |
| for callback in self._callbacks: |
| try: |
| if asyncio.iscoroutinefunction(callback): |
| await callback() |
| else: |
| callback() |
| except Exception as e: |
| logger.error(f"同步回调执行失败: {e}") |
| |
| async def start(self): |
| """启动同步""" |
| if self._running: |
| return |
| |
| self._running = True |
| await self._initial_sync() |
| |
| self._sync_task = asyncio.create_task(self._sync_loop()) |
| logger.info(f"双向同步已启动，间隔 {self._sync_interval} 秒") |
| |
| async def stop(self): |
| """停止同步""" |
| self._running = False |
| if self._sync_task: |
| self._sync_task.cancel() |
| try: |
| await self._sync_task |
| except asyncio.CancelledError: |
| pass |
| logger.info("双向同步已停止") |
| |
| async def _initial_sync(self): |
| """初始同步：合并两端数据""" |
| async with self._sync_lock: |
| try: |
| local_accounts = await self.local_db.load_accounts() |
| remote_accounts = await self.remote_db.load_accounts() |
| |
| merged = _merge_accounts_bidirectional(local_accounts, remote_accounts) |
| |
| await self.local_db.save_accounts(merged) |
| await self.remote_db.save_accounts(merged) |
| |
| self._local_hash = await self.local_db.get_config_hash() |
| self._remote_hash = await self.remote_db.get_config_hash() |
| |
| logger.info(f"初始同步完成，共 {len(merged)} 个账号") |
| await self._notify_callbacks() |
| |
| except Exception as e: |
| logger.error(f"初始同步失败: {e}") |
| |
| async def _sync_loop(self): |
| """同步循环""" |
| while self._running: |
| try: |
| await asyncio.sleep(self._sync_interval) |
| await self._check_and_sync() |
| except asyncio.CancelledError: |
| break |
| except Exception as e: |
| logger.error(f"同步循环异常: {e}") |
| |
| async def _check_and_sync(self): |
| """检查并同步变更""" |
| async with self._sync_lock: |
| try: |
| current_local_hash = await self.local_db.get_config_hash() |
| current_remote_hash = await self.remote_db.get_config_hash() |
| |
| local_changed = current_local_hash != self._local_hash |
| remote_changed = current_remote_hash != self._remote_hash |
| |
| if not local_changed and not remote_changed: |
| return |
| |
| local_accounts = await self.local_db.load_accounts() |
| remote_accounts = await self.remote_db.load_accounts() |
| |
| if local_changed and remote_changed: |
| logger.info("检测到本地和远程都有变更，执行合并") |
| merged = _merge_accounts_bidirectional(local_accounts, remote_accounts) |
| await self.local_db.save_accounts(merged) |
| await self.remote_db.save_accounts(merged) |
| |
| elif local_changed: |
| logger.info("检测到本地变更，同步到远程") |
| merged = _merge_accounts_bidirectional(local_accounts, remote_accounts) |
| await self.remote_db.save_accounts(merged) |
| await self.local_db.save_accounts(merged) |
| |
| elif remote_changed: |
| logger.info("检测到远程变更，同步到本地") |
| merged = _merge_accounts_bidirectional(local_accounts, remote_accounts) |
| await self.local_db.save_accounts(merged) |
| await self.remote_db.save_accounts(merged) |
| |
| self._local_hash = await self.local_db.get_config_hash() |
| self._remote_hash = await self.remote_db.get_config_hash() |
| |
| await self._notify_callbacks() |
| |
| except Exception as e: |
| logger.error(f"同步检查失败: {e}") |
| |
| async def force_sync(self, source: str = "merge"): |
| """强制同步 |
| |
| Args: |
| source: 同步源 - "local" (本地覆盖远程), "remote" (远程覆盖本地), "merge" (合并) |
| """ |
| async with self._sync_lock: |
| try: |
| if source == "local": |
| accounts = await self.local_db.load_accounts() |
| await self.remote_db.save_accounts(accounts) |
| logger.info(f"强制同步：本地 -> 远程，{len(accounts)} 个账号") |
| |
| elif source == "remote": |
| accounts = await self.remote_db.load_accounts() |
| await self.local_db.save_accounts(accounts) |
| logger.info(f"强制同步：远程 -> 本地，{len(accounts)} 个账号") |
| |
| else: |
| local_accounts = await self.local_db.load_accounts() |
| remote_accounts = await self.remote_db.load_accounts() |
| merged = _merge_accounts_bidirectional(local_accounts, remote_accounts) |
| await self.local_db.save_accounts(merged) |
| await self.remote_db.save_accounts(merged) |
| logger.info(f"强制同步：双向合并，{len(merged)} 个账号") |
| |
| self._local_hash = await self.local_db.get_config_hash() |
| self._remote_hash = await self.remote_db.get_config_hash() |
| |
| await self._notify_callbacks() |
| return True |
| |
| except Exception as e: |
| logger.error(f"强制同步失败: {e}") |
| return False |
| |
| async def get_sync_status(self) -> Dict[str, Any]: |
| """获取同步状态""" |
| try: |
| local_accounts = await self.local_db.load_accounts() |
| remote_accounts = await self.remote_db.load_accounts() |
| |
| current_local_hash = await self.local_db.get_config_hash() |
| current_remote_hash = await self.remote_db.get_config_hash() |
| |
| return { |
| "enabled": True, |
| "running": self._running, |
| "sync_interval": self._sync_interval, |
| "local_accounts": len(local_accounts), |
| "remote_accounts": len(remote_accounts), |
| "local_changed": current_local_hash != self._local_hash, |
| "remote_changed": current_remote_hash != self._remote_hash, |
| "in_sync": current_local_hash == current_remote_hash, |
| } |
| except Exception as e: |
| return { |
| "enabled": True, |
| "running": self._running, |
| "error": str(e), |
| } |
|
|
|
|
| def _merge_accounts_bidirectional( |
| local_accounts: List[Dict[str, Any]], |
| remote_accounts: List[Dict[str, Any]] |
| ) -> List[Dict[str, Any]]: |
| """双向合并账号，基于更新时间或合并策略""" |
| accounts_map: Dict[str, Dict[str, Any]] = {} |
| |
| for account in local_accounts: |
| key = _get_account_key(account) |
| account["_source"] = "local" |
| accounts_map[key] = account |
| |
| for account in remote_accounts: |
| key = _get_account_key(account) |
| if key in accounts_map: |
| local_acc = accounts_map[key] |
| local_updated = local_acc.get("updated_at", 0) |
| remote_updated = account.get("updated_at", 0) |
| |
| if remote_updated > local_updated: |
| account["_source"] = "remote" |
| accounts_map[key] = account |
| else: |
| account["_source"] = "remote" |
| accounts_map[key] = account |
| |
| merged = [] |
| for account in accounts_map.values(): |
| clean_account = {k: v for k, v in account.items() if not k.startswith("_")} |
| merged.append(clean_account) |
| |
| return merged |
|
|
|
|
| _sync_manager: Optional[SyncManager] = None |
|
|
|
|
| async def get_sync_manager() -> Optional[SyncManager]: |
| """获取同步管理器实例""" |
| return _sync_manager |
|
|
|
|
| async def start_sync_manager(): |
| """启动同步管理器""" |
| global _sync_manager |
| |
| database_url = os.getenv("DATABASE_URL") |
| if not database_url: |
| logger.info("未配置 DATABASE_URL，跳过同步管理器") |
| return |
| |
| from ..config import DATA_DIR |
| |
| local_db = FileSystemDatabase(DATA_DIR) |
| await local_db.initialize() |
| |
| remote_db = SQLDatabase(database_url) |
| if remote_db._db_type == "unknown": |
| logger.warning("DATABASE_URL 格式不支持，跳过同步管理器") |
| return |
|
|
| ok = await remote_db.initialize() |
| if not ok: |
| logger.warning("远程数据库初始化失败，跳过同步管理器") |
| return |
| |
| _sync_manager = SyncManager(local_db, remote_db) |
| await _sync_manager.start() |
|
|
|
|
| async def stop_sync_manager(): |
| """停止同步管理器""" |
| global _sync_manager |
| if _sync_manager: |
| await _sync_manager.stop() |
| _sync_manager = None |
|
|
