import requests import threading import time from local_cache import LocalFeatureCache import numpy as np class SyncManager: def __init__(self, cache: LocalFeatureCache, server_url, base_id, sync_interval=600): self.cache = cache self.server_url = server_url self.base_id = base_id self.interval = sync_interval # 秒 self.running = True self.thread = threading.Thread(target=self._sync_loop, daemon=True) self.thread.start() def _sync_loop(self): while self.running: try: self.download_incremental() self.upload_pending() except Exception as e: print(f"同步失败: {e}") time.sleep(self.interval) def download_incremental(self): last_time = self.cache.get_last_sync_time() resp = requests.get(f"{self.server_url}/api/templates/incremental", params={"since": last_time, "base_id": self.base_id}) if resp.status_code == 200: data = resp.json() for item in data['templates']: # 插入或更新 template_cache (使用REPLACE) feature_blob = np.array(item['feature'], dtype=np.float32).tobytes() self.cache.conn.execute( "INSERT OR REPLACE INTO template_cache (id, category, feature_blob, server_version) VALUES (?, ?, ?, ?)", (item['id'], item['category'], feature_blob, item['update_time']) ) if data.get('server_time'): self.cache.update_last_sync_time(data['server_time']) self.cache.conn.commit() def upload_pending(self): pendings = self.cache.get_pending_uploads() for p in pendings: payload = { "base_id": self.base_id, "category": p['category'], "feature": p['feature'].tolist(), "image_path": p['image_path'] if p['image_path'] else "" } resp = requests.post(f"{self.server_url}/api/templates/upload", json=payload) if resp.status_code == 200: self.cache.mark_uploaded(p['id']) else: # 失败则重试计数,稍后重试 self.cache.conn.execute( "UPDATE upload_queue SET retry_count=retry_count+1 WHERE id=?", (p['id'],) ) self.cache.conn.commit()