42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
"""
|
|
数据同步抽象基类
|
|
"""
|
|
import abc
|
|
from datetime import datetime
|
|
from core.logger import LogLevel, PrintLog
|
|
|
|
|
|
class BaseSync(abc.ABC):
|
|
"""数据同步抽象基类，所有同步操作遵循 _fetch → _upsert 模式"""
|
|
|
|
def __init__(self):
|
|
self.stats = {'inserted': 0, 'updated': 0, 'skipped': 0, 'errors': 0}
|
|
|
|
def run(self, **kwargs) -> dict:
|
|
"""同步入口: 拉取数据 → 写入数据库 → 返回统计"""
|
|
name = self.__class__.__name__
|
|
PrintLog(LogLevel.INFO, f'[sync] {name} 开始同步...')
|
|
t0 = datetime.now()
|
|
try:
|
|
data = self._fetch(**kwargs)
|
|
self._upsert(data)
|
|
elapsed = (datetime.now() - t0).total_seconds()
|
|
PrintLog(
|
|
LogLevel.INFO,
|
|
f'[sync] {name} 完成 ({elapsed:.1f}s) — '
|
|
f'insert={self.stats["inserted"]} update={self.stats["updated"]} '
|
|
f'skip={self.stats["skipped"]} err={self.stats["errors"]}'
|
|
)
|
|
except Exception as e:
|
|
PrintLog(LogLevel.ERROR, f'[sync] {name} 失败: {e}')
|
|
raise
|
|
return self.stats
|
|
|
|
@abc.abstractmethod
|
|
def _fetch(self, **kwargs):
|
|
"""从数据源拉取原始数据。子类实现。"""
|
|
|
|
@abc.abstractmethod
|
|
def _upsert(self, data):
|
|
"""将数据写入数据库。子类实现。"""
|