# coding=utf-8 """MySQL 总服务器 <-> SQLite 本地缓存 同步。 架构： - MySQL：总服务器主库（权威数据源） - SQLite：本地临时缓存（识别只读此库） - 变更流程：先写 MySQL → 再从 MySQL 拉取刷新 SQLite - 离线时：写入 SQLite + 入队，恢复后上传再全量拉取 """ import datetime import json import threading import time import numpy as np from sqlalchemy import func from dao.database import is_server_available, local_session_scope, server_session_scope from dao.feature_codec import feature_to_blob from dao.settings import DISH_SYNC_INTERVAL, DISH_SYNC_ON_START from models.App_settings import AppSetting from models.Bases import BaseInfo from models.Dish_categories import DishCategory from models.Global_templates import GlobalTemplate from models.Upload_queue import UploadQueue SETTING_LAST_SYNC = 'last_server_sync_time' SETTING_LAST_SYNC_ERROR = 'last_sync_error' def _soft_delete_category_name(category): marker = f'__del__{category.id}' if marker in (category.name or ''): return category.name base = (category.name or 'unnamed')[:80] return f'{base}{marker}' class TemplateSync: def __init__(self, source_base_id=None): self.source_base_id = source_base_id self.lock = threading.Lock() self._bg_thread = None self._running = False self._on_sync_callback = None self._migrate_legacy_upload_queue() def _migrate_legacy_upload_queue(self): with local_session_scope() as session: legacy = session.get(AppSetting, 'upload_queue_json') if not legacy or not legacy.value or legacy.value == '[]': return try: items = json.loads(legacy.value) except Exception: items = [] for item in items: session.add(UploadQueue( action=item.get('action', 'insert'), payload_json=json.dumps(item.get('payload', {}), ensure_ascii=False), status='pending', )) legacy.value = '[]' def _list_pending_uploads(self, session): rows = ( session.query(UploadQueue) .filter(UploadQueue.status == 'pending') .order_by(UploadQueue.id.asc()) .all() ) result = [] for row in rows: try: payload = json.loads(row.payload_json) if row.payload_json else {} except Exception: payload = {} result.append({'id': row.id, 'action': row.action, 'payload': payload}) return result def _get_last_sync_time(self, session): row = session.get(AppSetting, SETTING_LAST_SYNC) return row.value if row and row.value else '1970-01-01 00:00:00' def _set_last_sync_time(self, session, value): row = session.get(AppSetting, SETTING_LAST_SYNC) if row: row.value = value else: session.add(AppSetting(key=SETTING_LAST_SYNC, value=value)) def _set_last_sync_error(self, message): with local_session_scope() as session: row = session.get(AppSetting, SETTING_LAST_SYNC_ERROR) value = (message or '')[:500] if row: row.value = value else: session.add(AppSetting(key=SETTING_LAST_SYNC_ERROR, value=value)) def get_last_sync_time(self): with local_session_scope() as session: return self._get_last_sync_time(session) def get_last_sync_error(self): with local_session_scope() as session: row = session.get(AppSetting, SETTING_LAST_SYNC_ERROR) return (row.value if row and row.value else '') or '' def get_queue_stats(self): with local_session_scope() as session: pending = ( session.query(func.count(UploadQueue.id)) .filter(UploadQueue.status == 'pending') .scalar() ) or 0 failed = ( session.query(func.count(UploadQueue.id)) .filter(UploadQueue.status == 'failed') .scalar() ) or 0 return {'pending': int(pending), 'failed': int(failed)} def list_upload_queue(self, statuses=('pending', 'failed'), limit=100): statuses = tuple(statuses) if statuses else ('pending', 'failed') with local_session_scope() as session: rows = ( session.query(UploadQueue) .filter(UploadQueue.status.in_(statuses)) .order_by(UploadQueue.id.desc()) .limit(limit) .all() ) return [row.to_dict() for row in rows] def retry_failed_uploads(self): """将 failed 重置为 pending，并尝试 flush。返回处理结果摘要。""" with local_session_scope() as session: rows = session.query(UploadQueue).filter(UploadQueue.status == 'failed').all() reset = 0 for row in rows: row.status = 'pending' row.retry_count = 0 row.error_msg = None row.update_time = datetime.datetime.now() reset += 1 if reset == 0 and not is_server_available(): return {'ok': False, 'reset': 0, 'message': '无失败项，且总库离线'} if not is_server_available(): return {'ok': False, 'reset': reset, 'message': f'已重置 {reset} 条为待上传，总库离线稍后自动重试'} ok = self.flush_upload_queue() stats = self.get_queue_stats() return { 'ok': bool(ok), 'reset': reset, 'pending': stats['pending'], 'failed': stats['failed'], 'message': f'已重置 {reset} 条并尝试上传' if ok else '上传未完成，请检查总库', } def sync_now(self, full=True): """手动同步：先刷上传队列，再从总库拉取。""" if not is_server_available(): stats = self.get_queue_stats() self._set_last_sync_error('总库离线，无法同步') return { 'ok': False, 'online': False, 'pending': stats['pending'], 'failed': stats['failed'], 'last_sync': self.get_last_sync_time(), 'message': '总库离线，无法同步；本地变更已在上传队列中', } try: with local_session_scope() as session: had_pending = bool(self._list_pending_uploads(session)) if had_pending: # flush 成功时内部会 full pull self.flush_upload_queue() else: self.pull_from_server(full=bool(full)) stats = self.get_queue_stats() self._set_last_sync_error('') return { 'ok': True, 'online': True, 'pending': stats['pending'], 'failed': stats['failed'], 'last_sync': self.get_last_sync_time(), 'message': '同步完成' + (f'（仍有 {stats["failed"]} 条失败）' if stats['failed'] else ''), } except Exception as exc: self._set_last_sync_error(str(exc)) stats = self.get_queue_stats() return { 'ok': False, 'online': True, 'pending': stats['pending'], 'failed': stats['failed'], 'last_sync': self.get_last_sync_time(), 'message': f'同步失败：{exc}', } def get_sync_status(self): online = is_server_available() stats = self.get_queue_stats() return { 'online': online, 'last_sync': self.get_last_sync_time(), 'last_error': self.get_last_sync_error(), 'pending': stats['pending'], 'failed': stats['failed'], } def _get_or_create_category(self, session, name): category = ( session.query(DishCategory) .filter(DishCategory.name == name, DishCategory.is_active == 1) .first() ) if category: return category category = DishCategory(name=name, is_active=1) session.add(category) session.flush() return category @staticmethod def _category_to_dict(c): return { 'id': c.id, 'name': c.name, 'alias': c.alias, 'is_active': c.is_active, 'create_time': c.create_time, 'update_time': c.update_time, } @staticmethod def _template_to_dict(t): return { 'id': t.id, 'category_id': t.category_id, 'feature_blob': t.feature_blob, 'image_url': t.image_url, 'thumbnail_url': t.thumbnail_url, 'source_base_id': t.source_base_id, 'confidence': t.confidence, 'is_active': t.is_active, 'create_time': t.create_time, 'update_time': t.update_time, } @staticmethod def _base_to_dict(b): return { 'id': b.id, 'external_id': getattr(b, 'external_id', None), 'base_code': b.base_code, 'base_name': b.base_name, 'branch_name': getattr(b, 'branch_name', None), 'region': b.region, 'province': getattr(b, 'province', None), 'city': getattr(b, 'city', None), 'district': getattr(b, 'district', None), 'address': getattr(b, 'address', None), 'contact': b.contact, 'phone': getattr(b, 'phone', None), 'description': getattr(b, 'description', None), 'longitude': getattr(b, 'longitude', None), 'latitude': getattr(b, 'latitude', None), 'sort_order': getattr(b, 'sort_order', 0) or 0, 'is_active': b.is_active, 'register_time': b.register_time, 'last_heartbeat': b.last_heartbeat, } def _upsert_base_dict(self, local_session, data): existing = local_session.get(BaseInfo, data['id']) extra = { 'external_id': data.get('external_id'), 'branch_name': data.get('branch_name'), 'province': data.get('province'), 'city': data.get('city'), 'district': data.get('district'), 'address': data.get('address'), 'phone': data.get('phone'), 'description': data.get('description'), 'longitude': data.get('longitude'), 'latitude': data.get('latitude'), 'sort_order': data.get('sort_order') or 0, } if existing: existing.base_code = data['base_code'] existing.base_name = data['base_name'] existing.region = data.get('region') existing.contact = data.get('contact') existing.is_active = data.get('is_active', 1) existing.register_time = data.get('register_time') or existing.register_time existing.last_heartbeat = data.get('last_heartbeat') or existing.last_heartbeat for k, v in extra.items(): if hasattr(existing, k): setattr(existing, k, v) else: row = BaseInfo( id=data['id'], base_code=data['base_code'], base_name=data['base_name'], region=data.get('region'), contact=data.get('contact'), is_active=data.get('is_active', 1), register_time=data.get('register_time') or datetime.datetime.now(), last_heartbeat=data.get('last_heartbeat') or datetime.datetime.now(), ) for k, v in extra.items(): if hasattr(row, k): setattr(row, k, v) local_session.add(row) def _upsert_category_dict(self, local_session, data): existing = local_session.get(DishCategory, data['id']) if existing: existing.name = data['name'] existing.alias = data['alias'] existing.is_active = data['is_active'] existing.create_time = data['create_time'] existing.update_time = data['update_time'] else: local_session.add(DishCategory( id=data['id'], name=data['name'], alias=data['alias'], is_active=data['is_active'], create_time=data['create_time'], update_time=data['update_time'], )) def _upsert_template_dict(self, local_session, data): existing = local_session.get(GlobalTemplate, data['id']) if existing: existing.category_id = data['category_id'] existing.feature_blob = data['feature_blob'] existing.image_url = data.get('image_url') existing.thumbnail_url = data.get('thumbnail_url') existing.source_base_id = data['source_base_id'] existing.confidence = data['confidence'] existing.is_active = data['is_active'] existing.create_time = data['create_time'] existing.update_time = data['update_time'] else: local_session.add(GlobalTemplate( id=data['id'], category_id=data['category_id'], feature_blob=data['feature_blob'], image_url=data.get('image_url'), thumbnail_url=data.get('thumbnail_url'), source_base_id=data['source_base_id'], confidence=data['confidence'], is_active=data['is_active'], create_time=data['create_time'], update_time=data['update_time'], )) def _replace_local_from_server_dicts(self, categories, templates, threshold_value, sync_time, bases=None): with local_session_scope() as local_session: local_session.query(GlobalTemplate).delete() local_session.query(DishCategory).delete() if bases is not None: local_session.query(BaseInfo).delete() local_session.flush() for data in (bases or []): row = BaseInfo( id=data['id'], base_code=data['base_code'], base_name=data['base_name'], region=data.get('region'), contact=data.get('contact'), is_active=data.get('is_active', 1), register_time=data.get('register_time') or datetime.datetime.now(), last_heartbeat=data.get('last_heartbeat') or datetime.datetime.now(), ) for k in ( 'external_id', 'branch_name', 'province', 'city', 'district', 'address', 'phone', 'description', 'longitude', 'latitude', 'sort_order', ): if hasattr(row, k): setattr(row, k, data.get(k) if k != 'sort_order' else (data.get(k) or 0)) local_session.add(row) for data in categories: local_session.add(DishCategory( id=data['id'], name=data['name'], alias=data['alias'], is_active=data['is_active'], create_time=data['create_time'], update_time=data['update_time'], )) for data in templates: local_session.add(GlobalTemplate( id=data['id'], category_id=data['category_id'], feature_blob=data['feature_blob'], image_url=data.get('image_url'), thumbnail_url=data.get('thumbnail_url'), source_base_id=data['source_base_id'], confidence=data['confidence'], is_active=data['is_active'], create_time=data['create_time'], update_time=data['update_time'], )) if threshold_value is not None: row = local_session.get(AppSetting, 'similarity_threshold') if row: row.value = threshold_value else: local_session.add(AppSetting(key='similarity_threshold', value=threshold_value)) self._set_last_sync_time(local_session, sync_time) def pull_from_server(self, full=False): if not is_server_available(): return False with self.lock: with local_session_scope() as local_session: last_sync = self._get_last_sync_time(local_session) local_count = local_session.query(func.count(GlobalTemplate.id)).scalar() or 0 use_full = full or local_count == 0 or last_sync == '1970-01-01 00:00:00' with server_session_scope() as server_session: bases = [self._base_to_dict(b) for b in server_session.query(BaseInfo).all()] if use_full: categories = [self._category_to_dict(c) for c in server_session.query(DishCategory).all()] templates = [self._template_to_dict(t) for t in server_session.query(GlobalTemplate).all()] else: since = datetime.datetime.strptime(last_sync, '%Y-%m-%d %H:%M:%S') categories = [ self._category_to_dict(c) for c in server_session.query(DishCategory).filter(DishCategory.update_time >= since).all() ] templates = [ self._template_to_dict(t) for t in server_session.query(GlobalTemplate).filter(GlobalTemplate.update_time >= since).all() ] thr = server_session.get(AppSetting, 'similarity_threshold') threshold_value = thr.value if thr else None max_update = datetime.datetime(1970, 1, 1) for item in categories + templates: ut = item.get('update_time') if ut and ut > max_update: max_update = ut sync_time = ( max_update.strftime('%Y-%m-%d %H:%M:%S') if max_update.year > 1970 else datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') ) if use_full: self._replace_local_from_server_dicts( categories, templates, threshold_value, sync_time, bases=bases, ) else: with local_session_scope() as local_session: for data in bases: self._upsert_base_dict(local_session, data) for data in categories: self._upsert_category_dict(local_session, data) for data in templates: self._upsert_template_dict(local_session, data) if threshold_value is not None: row = local_session.get(AppSetting, 'similarity_threshold') if row: row.value = threshold_value else: local_session.add(AppSetting(key='similarity_threshold', value=threshold_value)) self._set_last_sync_time(local_session, sync_time) try: from dao.base_registry import touch_base_heartbeat touch_base_heartbeat(self.source_base_id) except Exception: pass mode = 'full' if use_full else 'incremental' print(f"[OK] pulled templates from mysql to sqlite ({mode})") return True def flush_upload_queue(self): if not is_server_available(): return False with self.lock: with local_session_scope() as local_session: queue = self._list_pending_uploads(local_session) if not queue: return True for item in queue: try: action = item.get('action') if action == 'insert': self._server_insert(item['payload'], refresh=False) elif action == 'delete_template': self._server_delete_template(item['payload']['template_id'], refresh=False) elif action == 'delete_category': self._server_delete_category(item['payload']['category_name'], refresh=False) elif action == 'clear_all': self._server_clear_all(refresh=False) elif action == 'set_threshold': self._server_set_threshold(item['payload']['threshold'], refresh=False) with local_session_scope() as local_session: row = local_session.get(UploadQueue, item['id']) if row: row.status = 'done' row.update_time = datetime.datetime.now() except Exception as exc: print(f"[WARN] upload queue item failed: {exc}") with local_session_scope() as local_session: row = local_session.get(UploadQueue, item['id']) if row: row.retry_count = (row.retry_count or 0) + 1 row.error_msg = str(exc) row.update_time = datetime.datetime.now() if row.retry_count >= 5: row.status = 'failed' self.pull_from_server(full=True) return True def _enqueue(self, action, payload): with local_session_scope() as session: session.add(UploadQueue( action=action, payload_json=json.dumps(payload, ensure_ascii=False), status='pending', )) def _server_insert(self, payload, refresh=True): if not is_server_available(): self._enqueue('insert', payload) self._insert_local(payload) return None with server_session_scope() as session: category = self._get_or_create_category(session, payload['category_name']) template = GlobalTemplate( category_id=category.id, feature_blob=feature_to_blob(payload['feature']), image_url=payload.get('image_url') or payload.get('image_path'), thumbnail_url=payload.get('thumbnail_url') or payload.get('thumbnail_path'), source_base_id=self.source_base_id, confidence=1.0, is_active=1, ) session.add(template) session.flush() template_id = template.id if refresh: self.pull_from_server(full=False) return template_id def _deactivate_category(self, session, category_name): category = ( session.query(DishCategory) .filter(DishCategory.name == category_name, DishCategory.is_active == 1) .first() ) if not category: return now = datetime.datetime.now() for template in session.query(GlobalTemplate).filter( GlobalTemplate.category_id == category.id, GlobalTemplate.is_active == 1, ): template.is_active = 0 template.update_time = now category.is_active = 0 category.name = _soft_delete_category_name(category) category.update_time = now def _server_delete_template(self, template_id, refresh=True): if not is_server_available(): self._enqueue('delete_template', {'template_id': template_id}) self._delete_template_local(template_id) return with server_session_scope() as session: template = session.get(GlobalTemplate, template_id) if template and template.is_active == 1: template.is_active = 0 template.update_time = datetime.datetime.now() if refresh: self.pull_from_server(full=False) def _server_delete_category(self, category_name, refresh=True): if not is_server_available(): self._enqueue('delete_category', {'category_name': category_name}) self._delete_category_local(category_name) return with server_session_scope() as session: self._deactivate_category(session, category_name) if refresh: self.pull_from_server(full=False) def _server_clear_all(self, refresh=True): if not is_server_available(): self._enqueue('clear_all', {}) self._clear_all_local() return with server_session_scope() as session: now = datetime.datetime.now() for template in session.query(GlobalTemplate).filter(GlobalTemplate.is_active == 1): template.is_active = 0 template.update_time = now for category in session.query(DishCategory).filter(DishCategory.is_active == 1): category.is_active = 0 category.name = _soft_delete_category_name(category) category.update_time = now if refresh: self.pull_from_server(full=True) def _server_set_threshold(self, threshold, refresh=True): value = str(float(threshold)) if not is_server_available(): self._enqueue('set_threshold', {'threshold': threshold}) self._set_threshold_local(threshold) return with server_session_scope() as session: row = session.get(AppSetting, 'similarity_threshold') if row: row.value = value else: session.add(AppSetting(key='similarity_threshold', value=value)) with local_session_scope() as session: row = session.get(AppSetting, 'similarity_threshold') if row: row.value = value else: session.add(AppSetting(key='similarity_threshold', value=value)) if refresh: self.pull_from_server(full=False) def push_insert(self, category_name, image_path, feature, thumbnail_path): payload = { 'category_name': category_name, 'image_url': image_path, 'feature': np.asarray(feature, dtype=np.float32).tolist(), 'thumbnail_url': thumbnail_path, } return self._server_insert(payload, refresh=True) def push_delete_template(self, template_id): self._server_delete_template(template_id, refresh=True) def push_delete_category(self, category_name): self._server_delete_category(category_name, refresh=True) def push_clear_all(self): self._server_clear_all(refresh=True) def push_set_threshold(self, threshold): self._server_set_threshold(threshold, refresh=True) def push_insert_batch(self, items): if not items: return 0 if not is_server_available(): for item in items: payload = { 'category_name': item['category_name'], 'image_url': item.get('image_url') or item.get('image_path'), 'feature': item['feature'], 'thumbnail_url': item.get('thumbnail_url') or item.get('thumbnail_path'), } self._enqueue('insert', payload) self._insert_local(payload) return len(items) with server_session_scope() as session: for item in items: category = self._get_or_create_category(session, item['category_name']) session.add(GlobalTemplate( category_id=category.id, feature_blob=feature_to_blob(item['feature']), image_url=item.get('image_url') or item.get('image_path'), thumbnail_url=item.get('thumbnail_url') or item.get('thumbnail_path'), source_base_id=self.source_base_id, confidence=1.0, is_active=1, )) self.pull_from_server(full=True) return len(items) def _insert_local(self, payload): with local_session_scope() as session: category = self._get_or_create_category(session, payload['category_name']) template = GlobalTemplate( category_id=category.id, feature_blob=feature_to_blob(payload['feature']), image_url=payload.get('image_url') or payload.get('image_path'), thumbnail_url=payload.get('thumbnail_url') or payload.get('thumbnail_path'), source_base_id=self.source_base_id, confidence=1.0, is_active=1, ) session.add(template) session.flush() return template.id def _delete_template_local(self, template_id): with local_session_scope() as session: template = session.get(GlobalTemplate, template_id) if template and template.is_active == 1: template.is_active = 0 template.update_time = datetime.datetime.now() return {'path': template.image_url, 'thumbnail': template.thumbnail_url} return None def _delete_category_local(self, category_name): removed = [] with local_session_scope() as session: category = ( session.query(DishCategory) .filter(DishCategory.name == category_name, DishCategory.is_active == 1) .first() ) if not category: return removed now = datetime.datetime.now() templates = ( session.query(GlobalTemplate) .filter(GlobalTemplate.category_id == category.id, GlobalTemplate.is_active == 1) .all() ) for template in templates: template.is_active = 0 template.update_time = now removed.append({'path': template.image_url, 'thumbnail': template.thumbnail_url}) category.is_active = 0 category.name = _soft_delete_category_name(category) category.update_time = now return removed def _clear_all_local(self): with local_session_scope() as session: now = datetime.datetime.now() for template in session.query(GlobalTemplate).filter(GlobalTemplate.is_active == 1): template.is_active = 0 template.update_time = now for category in session.query(DishCategory).filter(DishCategory.is_active == 1): category.is_active = 0 category.name = _soft_delete_category_name(category) category.update_time = now def _set_threshold_local(self, threshold): with local_session_scope() as session: row = session.get(AppSetting, 'similarity_threshold') value = str(float(threshold)) if row: row.value = value else: session.add(AppSetting(key='similarity_threshold', value=value)) def initial_sync(self): if not is_server_available(): print("[WARN] server unavailable, using sqlite cache only") return False print("[INFO] startup: flush upload queue…") self.flush_upload_queue() if not DISH_SYNC_ON_START: print("[INFO] DISH_SYNC_ON_START=0, skip template pull") return True with local_session_scope() as local_session: local_count = local_session.query(func.count(GlobalTemplate.id)).scalar() or 0 # 本地已有模板时不要启动全量拉取（大库会卡死数分钟且无日志） if local_count > 0: print( f"[INFO] local templates={local_count}, " "startup incremental sync (skip full pull)…" ) self.pull_from_server(full=False) else: print("[INFO] local templates empty, full pull from mysql (may take a while)…") self.pull_from_server(full=True) return True def start_background_sync(self, interval=None, on_sync=None): if not is_server_available(): return self._on_sync_callback = on_sync self._running = True wait = interval or DISH_SYNC_INTERVAL def loop(): while self._running: time.sleep(wait) try: # 后台也刷队列，避免离线变更一直积压 with local_session_scope() as session: had_pending = bool(self._list_pending_uploads(session)) if had_pending: self.flush_upload_queue() else: self.pull_from_server(full=False) self._set_last_sync_error('') if self._on_sync_callback: self._on_sync_callback() except Exception as exc: print(f"[WARN] background sync failed: {exc}") try: self._set_last_sync_error(str(exc)) except Exception: pass self._bg_thread = threading.Thread(target=loop, daemon=True) self._bg_thread.start() def stop_background_sync(self): self._running = False