Update r2_sync.py

r2_sync.py

@@ -17,10 +17,10 @@ from urllib.parse import unquote
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - %(message)s',
-    # handlers=[
-    #     logging.FileHandler('r2_sync.log'),
-    #     logging.StreamHandler()
-    # ]
+    handlers=[
+        logging.FileHandler('r2_sync.log'),
+        logging.StreamHandler()
+    ]
 )
 logger = logging.getLogger(__name__)
 
@@ -35,24 +35,24 @@ class Config:
         self.max_upload_workers = int(os.getenv("MAX_UPLOAD_WORKERS", "4"))
         self.upload_retry_delay = int(os.getenv("UPLOAD_RETRY_DELAY", "15"))  # 秒
         
-        # 忽略的文件后缀配置（逗号分隔，如 ".tmp,.log"）
-        ignore_suffixes = os.getenv("IGNORE_SUFFIXES", ".db-journal,.tmp,.log")
-        self.ignore_suffixes = [s.strip().lower() for s in ignore_suffixes.split(",") if s.strip()]
+        # 默认忽略的临时文件后缀
+        self.ignore_suffixes = [".tmp", ".log", ".bak", ".swp", ".db-journal"]
+        # 从环境变量追加忽略后缀
+        custom_ignore = os.getenv("IGNORE_SUFFIXES", "")
+        self.ignore_suffixes.extend([s.strip().lower() for s in custom_ignore.split(",") if s.strip()])
         
-        # 只同步的文件后缀配置（逗号分隔，如 ".jpg,.png"，优先级高于忽略后缀）
+        # 白名单模式（优先级高于黑名单）
         only_suffixes = os.getenv("ONLY_SUFFIXES", "")
         self.only_suffixes = [s.strip().lower() for s in only_suffixes.split(",") if s.strip()]
         
-        # 验证配置
         if not all([self.api_key, self.api_endpoint]):
             raise ValueError("必须设置API_AUTH_KEY和API_ENDPOINT环境变量")
         
-        # 检查冲突配置
-        if self.ignore_suffixes and self.only_suffixes:
-            logger.warning("同时设置了IGNORE_SUFFIXES和ONLY_SUFFIXES，将优先使用ONLY_SUFFIXES")
+        if self.only_suffixes and self.ignore_suffixes:
+            logger.info(f"白名单模式激活，将只同步以下后缀的文件: {', '.join(self.only_suffixes)}")
 
 class SyncHandler(FileSystemEventHandler):
-    """文件系统事件处理器"""
+    """文件系统事件处理器（带过滤功能）"""
     def __init__(self, callback, config):
         self.callback = callback
         self.config = config
@@ -61,16 +61,12 @@ class SyncHandler(FileSystemEventHandler):
         """检查文件是否应该被同步"""
         file_path = str(file_path).lower()
         
-        # 优先检查白名单模式
+        # 白名单模式优先
         if self.config.only_suffixes:
             return any(file_path.endswith(suffix) for suffix in self.config.only_suffixes)
         
-        # 然后检查黑名单模式
-        if self.config.ignore_suffixes:
-            return not any(file_path.endswith(suffix) for suffix in self.config.ignore_suffixes)
-        
-        # 默认同步所有文件
-        return True
+        # 黑名单模式
+        return not any(file_path.endswith(suffix) for suffix in self.config.ignore_suffixes)
     
     def on_modified(self, event):
         if not event.is_directory and self.should_sync(event.src_path):
@@ -81,37 +77,28 @@ class SyncHandler(FileSystemEventHandler):
             self.callback(event.src_path)
 
 class R2Sync:
-    """R2存储同步器"""
+    """R2存储同步器（优化版）"""
     def __init__(self, config):
         self.config = config
-        self.file_states = {}  # 跟踪文件最后状态 (size, mtime, hash?)
-        self.upload_queue = {}  # 上传队列 {file_path: (key, mtime, timer)}
-        self.remote_files_cache = None  # 缓存远程文件列表
-        self.cache_valid = False  # 缓存是否有效
+        self.file_states = {}  # {file_path: (size, hash?)}
+        self.upload_queue = {}  # {file_path: (key, size, hash?, timer)}
+        self.remote_files_cache = None
+        self.cache_valid = False
         self.executor = ThreadPoolExecutor(max_workers=self.config.max_upload_workers)
         self.cache_lock = threading.Lock()
         self.upload_queue_lock = threading.Lock()
         
-        # 确保数据目录存在
         self.config.data_dir.mkdir(parents=True, exist_ok=True)
     
     def should_sync_file(self, file_path):
-        """检查文件是否应该被同步"""
+        """统一文件过滤逻辑"""
         file_path = str(file_path).lower()
-        
-        # 优先检查白名单模式
         if self.config.only_suffixes:
             return any(file_path.endswith(suffix) for suffix in self.config.only_suffixes)
-        
-        # 然后检查黑名单模式
-        if self.config.ignore_suffixes:
-            return not any(file_path.endswith(suffix) for suffix in self.config.ignore_suffixes)
-        
-        # 默认同步所有文件
-        return True
+        return not any(file_path.endswith(suffix) for suffix in self.config.ignore_suffixes)
     
     def r2_api_request(self, method, path, data=None, max_retries=3):
-        """发送API请求"""
+        """带重试机制的API请求"""
         url = f"{self.config.api_endpoint}/{path}"
         headers = {
             'X-API-Key': self.config.api_key,
@@ -124,56 +111,53 @@ class R2Sync:
                     resp = requests.get(url, headers=headers)
                 elif method == "POST":
                     resp = requests.post(url, data=data, headers=headers)
-                elif method == "PUT":
-                    resp = requests.put(url, data=data, headers=headers)
                 elif method == "DELETE":
                     resp = requests.delete(url, headers=headers)
-                    
-                resp.raise_for_status()
                 
-                if method == "GET":
-                    return resp.content
-                elif method == "DELETE":
-                    return True
-                return resp.json()
+                resp.raise_for_status()
+                return resp.content if method == "GET" else resp.json()
                 
             except requests.exceptions.RequestException as e:
                 if attempt == max_retries - 1:
-                    logger.error(f"API请求最终失败: {e}")
+                    logger.error(f"API请求失败: {e}")
                     return None
                 wait_time = (attempt + 1) * 2
-                logger.warning(f"API请求失败(尝试 {attempt + 1}/{max_retries}), {wait_time}秒后重试...")
+                logger.warning(f"请求失败，{wait_time}秒后重试...")
                 time.sleep(wait_time)
     
     def get_file_hash(self, file_path):
-        """计算文件哈希值"""
+        """计算文件哈希（Blake2b算法）"""
+        if not self.config.use_hash_check:
+            return None
+            
         hash_obj = hashlib.blake2b()
         with open(file_path, 'rb') as f:
             while chunk := f.read(8192):
                 hash_obj.update(chunk)
         return hash_obj.hexdigest()
     
-    def is_file_modified(self, file_path, last_known_state):
-        """检查文件是否修改"""
+    def is_file_modified(self, file_path):
+        """优化版修改检测：仅当大小变化或哈希不匹配时返回True"""
         try:
-            stat = os.stat(file_path)
-            current_state = (stat.st_size, stat.st_mtime)
+            current_size = os.path.getsize(file_path)
+            last_state = self.file_states.get(file_path, (None, None))
             
-            if last_known_state[:2] == current_state:
-                return False
+            # 大小不同肯定被修改了
+            if current_size != last_state[0]:
+                return True
                 
+            # 大小相同但启用了哈希检查
             if self.config.use_hash_check:
                 current_hash = self.get_file_hash(file_path)
-                if len(last_known_state) > 2:
-                    return current_hash != last_known_state[2]
-                return True
-            return True
+                return current_hash != last_state[1]
+                
+            return False
         except Exception as e:
-            logger.error(f"检查文件修改状态出错: {e}")
+            logger.error(f"修改检查错误: {e}")
             return False
-    
+
     def get_remote_files(self, force_refresh=False):
-        """获取远程文件列表，使用缓存"""
+        """带缓存的远程文件列表获取"""
         with self.cache_lock:
             if force_refresh or not self.cache_valid or self.remote_files_cache is None:
                 self.remote_files_cache = self._fetch_remote_files()
@@ -187,195 +171,100 @@ class R2Sync:
             return None
             
         try:
-            raw_resp = resp.decode('utf-8').strip()
-            if raw_resp == "[]":
-                return []
-                
-            data = json.loads(raw_resp)
-            
-            # 统一处理为列表格式
+            data = json.loads(resp.decode('utf-8'))
             if isinstance(data, dict):
-                if 'objects' in data:
-                    data = data['objects']
-                else:
-                    data = [data]
-            elif not isinstance(data, list):
-                data = [data]
-                
-            # 处理文件名编码
-            for item in data:
-                if isinstance(item, dict) and 'key' in item:
-                    try:
-                        item['key'] = unquote(item['key'])
-                    except:
-                        pass
-                        
-            return data
+                return data.get('objects', [data])
+            return data if isinstance(data, list) else [data]
         except Exception as e:
-            logger.error(f"解析远程文件列表出错: {e}")
+            logger.error(f"解析文件列表出错: {e}")
             return []
-    
-    def invalidate_cache(self):
-        """使缓存失效"""
-        with self.cache_lock:
-            self.cache_valid = False
-    
+
     def download_file(self, key, dest_path):
-        """下载文件"""
+        """下载文件到本��"""
         if not self.should_sync_file(dest_path):
-            logger.debug(f"忽略下载文件（不匹配同步规则）: {key}")
+            logger.debug(f"忽略下载: {key}")
             return False
             
-        path = f"download/{key}"
-        logger.debug(f"下载路径: {dest_path} (原始key: {key})")
-        
-        resp = self.r2_api_request("GET", path)
-        if not resp:
+        content = self.r2_api_request("GET", f"download/{key}")
+        if not content:
             return False
             
-        # 确保目标目录存在
         dest_path.parent.mkdir(parents=True, exist_ok=True)
-        
-        # 处理不同响应类型
-        if isinstance(resp, bytes):
-            content = resp
-        else:
-            try:
-                content = resp.decode('utf-8').encode('utf-8')
-            except UnicodeDecodeError:
-                content = resp
-        
         with open(dest_path, 'wb') as f:
-            f.write(content)
+            f.write(content if isinstance(content, bytes) else content.encode('utf-8'))
         
-        logger.info(f"文件已保存到: {dest_path} (大小: {len(content)}字节)")
+        # 记录文件状态
+        file_size = os.path.getsize(dest_path)
+        file_hash = self.get_file_hash(dest_path) if self.config.use_hash_check else None
+        self.file_states[dest_path] = (file_size, file_hash)
+        
+        logger.info(f"下载完成: {key} ({file_size}字节)")
         return True
-    
-    def _upload_file_task(self, file_path, key):
-        """线程池中执行的上传任务"""
+
+    def upload_file(self, file_path, key):
+        """上传文件到R2"""
         if not self.should_sync_file(file_path):
-            logger.debug(f"忽略上传文件（不匹配同步规则）: {key}")
+            logger.debug(f"忽略上传: {key}")
             return False
             
-        path = f"upload/{key}"
         try:
             with open(file_path, 'rb') as f:
-                result = self.r2_api_request("POST", path, f.read()) is not None
-                if result:
-                    self.invalidate_cache()  # 上传成功，使缓存失效
-                return result
+                success = self.r2_api_request("POST", f"upload/{key}", f.read()) is not None
+                if success:
+                    self.invalidate_cache()
+                return success
         except Exception as e:
-            logger.error(f"上传文件出错: {e}")
+            logger.error(f"上传失败: {e}")
             return False
-    
-    def upload_file(self, file_path, key):
-        """上传文件"""
-        # 提交到线程池执行
-        future = self.executor.submit(self._upload_file_task, file_path, key)
-        return future.result()  # 阻塞等待结果
-    
+
     def delete_file(self, key):
-        """删除文件并更新缓存"""
+        """删除远程文件"""
         if not self.should_sync_file(key):
-            logger.debug(f"忽略删除文件（不匹配同步规则）: {key}")
+            logger.debug(f"忽略删除: {key}")
             return False
             
-        logger.info(f"删除远程文件: {key}")
-        path = f"delete/{key}"
-        if not self.r2_api_request("DELETE", path):
+        if not self.r2_api_request("DELETE", f"delete/{key}"):
             return False
             
-        # 使缓存失效
         self.invalidate_cache()
-        
-        # 验证文件是否真的被删除
-        remote_files = self.get_remote_files(force_refresh=True)
-        if remote_files is None:
-            return False
-            
-        return not any(file_info['key'] == key for file_info in remote_files)
-    
-    def check_api_health(self):
-        """检查API健康状态"""
-        logger.info("检查API健康状态...")
-        
-        # 测试列表API
-        files = self.get_remote_files()
-        if files is None:
-            logger.error("列表API测试失败")
-            return False
-            
-        logger.info("API健康状态检查通过")
         return True
-    
+
     def init_sync(self):
-        """初始化同步"""
-        logger.info("初始化数据目录...")
-        
-        if not self.check_api_health():
-            logger.error("API检查失败，请检查配置和网络连接")
-            sys.exit(1)
-        
-        # 获取远程文件列表
-        remote_files = self.get_remote_files()
-        if remote_files is None:
-            logger.error("获取远程文件列表失败")
+        """初始化同步：下载所有远程文件"""
+        if not self.get_remote_files():
+            logger.error("无法获取远程文件列表")
             sys.exit(1)
             
-        logger.info(f"找到 {len(remote_files)} 个远程文件")
-        if not remote_files:
-            logger.info("远程存储桶为空，无需同步")
-            return
-        
-        logger.info("开始同步远程文件...")
-        for file_info in remote_files:
+        for file_info in self.get_remote_files():
             key = file_info['key']
             dest_path = self.config.data_dir / key
-            
-            if not self.should_sync_file(dest_path):
-                logger.debug(f"忽略下载文件（不匹配同步规则）: {key}")
-                continue
-                
-            logger.info(f"下载: {key} -> {dest_path}")
-            if self.download_file(key, dest_path):
-                # 记录文件初始状态
-                stat = dest_path.stat()
-                file_state = (stat.st_size, stat.st_mtime)
-                if self.config.use_hash_check:
-                    file_state += (self.get_file_hash(dest_path),)
-                self.file_states[key] = file_state
-    
+            if self.should_sync_file(dest_path):
+                self.download_file(key, dest_path)
+
     def handle_file_change(self, file_path):
-        """处理文件变化事件，加入上传队列并设置定时检查"""
+        """处理文件变化事件（带防抖）"""
         try:
             if not self.should_sync_file(file_path):
-                logger.debug(f"忽略文件变化（不匹配同步规则）: {file_path}")
                 return
                 
-            rel_path = Path(file_path).relative_to(self.config.data_dir)
-            key = str(rel_path).replace('\\', '/')
-            
-            # 获取文件当前状态
-            stat = os.stat(file_path)
-            current_mtime = stat.st_mtime
+            key = str(Path(file_path).relative_to(self.config.data_dir)).replace('\\', '/')
+            current_size = os.path.getsize(file_path)
+            current_hash = self.get_file_hash(file_path) if self.config.use_hash_check else None
             
             with self.upload_queue_lock:
-                # 如果文件已在队列中，更新修改时间并重置计时器
+                # 取消已有计时器
                 if file_path in self.upload_queue:
-                    _, _, timer = self.upload_queue[file_path]
-                    timer.cancel()  # 取消之前的计时器
+                    self.upload_queue[file_path][3].cancel()
                 
-                # 创建新计时器，15秒后检查
+                # 创建新计时器
                 timer = threading.Timer(
                     self.config.upload_retry_delay,
                     self._process_upload_queue,
                     args=(file_path,)
                 )
                 timer.start()
-                
-                # 加入/更新队列
-                self.upload_queue[file_path] = (key, current_mtime, timer)
-                logger.info(f"检测到文件变化，加入上传队列: {key} (将在{self.config.upload_retry_delay}秒后检查)")
+                self.upload_queue[file_path] = (key, current_size, current_hash, timer)
+                logger.info(f"文件加入上传队列: {key}")
                 
         except Exception as e:
             logger.error(f"处理文件变化出错: {e}")
@@ -385,100 +274,57 @@ class R2Sync:
         with self.upload_queue_lock:
             if file_path not in self.upload_queue:
                 return
-                
-            key, original_mtime, _ = self.upload_queue[file_path]
-            del self.upload_queue[file_path]  # 从队列移除
-            
+            key, original_size, original_hash, _ = self.upload_queue.pop(file_path)
+        
         try:
-            # 检查文件是否还存在
             if not os.path.exists(file_path):
-                logger.info(f"文件已被删除，取消上传: {key}")
+                logger.info(f"文件已删除: {key}")
                 return
                 
-            # 获取当前文件状态
-            current_stat = os.stat(file_path)
-            current_mtime = current_stat.st_mtime
+            current_size = os.path.getsize(file_path)
+            current_hash = self.get_file_hash(file_path) if self.config.use_hash_check else None
             
-            # 比较修改时间，确认文件是否又有新变化
-            if current_mtime != original_mtime:
-                logger.info(f"文件 {key} 在等待期间又有新变化，重新加入队列")
-                self.handle_file_change(file_path)  # 重新加入队列
+            # 检查是否有新变化
+            if current_size != original_size or (
+                self.config.use_hash_check and current_hash != original_hash
+            ):
+                logger.info(f"文件有变化，重新加入队列: {key}")
+                self.handle_file_change(file_path)
                 return
                 
-            # 确认文件无新变化，开始上传
-            logger.info(f"开始上传文件: {key}")
+            # 执行上传
             if self.upload_file(file_path, key):
-                # 更新文件状态
-                file_state = (current_stat.st_size, current_mtime)
-                if self.config.use_hash_check:
-                    file_state += (self.get_file_hash(file_path),)
-                self.file_states[key] = file_state
-            else:
-                logger.error(f"上传失败: {key}")
-                
-        except Exception as e:
-            logger.error(f"处理上传队列出错: {e}")
-
-    def sync_deleted_files(self):
-        """同步删除操作"""
-        try:
-            remote_files = self.get_remote_files()
-            if remote_files is None:
-                return
-                
-            local_files = {
-                str(f.relative_to(self.config.data_dir)).replace('\\', '/')
-                for f in self.config.data_dir.rglob('*') 
-                if f.is_file() and self.should_sync_file(f)
-            }
-            
-            for file_info in remote_files:
-                key = file_info['key']
-                if key not in local_files and key not in self.upload_queue:
-                    logger.info(f"删除远程文件: {key}")
-                    if self.delete_file(key):
-                        if key in self.file_states:
-                            del self.file_states[key]
-                    else:
-                        logger.error(f"删除失败: {key}")
+                self.file_states[file_path] = (current_size, current_hash)
         except Exception as e:
-            logger.error(f"同步删除操作出错: {e}")
+            logger.error(f"处理上传出错: {e}")
 
     def watch_and_sync(self):
-        """监控并同步文件"""
-        logger.info("启动持续同步服务...")
-        
-        # 初始化文件状态（根据同步规则）
+        """主监控循环"""
+        # 初始化文件状态
         for file in self.config.data_dir.rglob('*'):
             if file.is_file() and self.should_sync_file(file):
-                rel_path = str(file.relative_to(self.config.data_dir)).replace('\\', '/')
-                stat = file.stat()
-                file_state = (stat.st_size, stat.st_mtime)
-                if self.config.use_hash_check:
-                    file_state += (self.get_file_hash(file),)
-                self.file_states[rel_path] = file_state
+                self.file_states[file] = (
+                    os.path.getsize(file),
+                    self.get_file_hash(file) if self.config.use_hash_check else None
+                )
         
-        # 设置文件监控（根据同步规则）
-        event_handler = SyncHandler(self.handle_file_change, self.config)
+        # 启动文件监控
         observer = Observer()
-        observer.schedule(event_handler, path=str(self.config.data_dir), recursive=True)
+        observer.schedule(
+            SyncHandler(self.handle_file_change, self.config),
+            path=str(self.config.data_dir),
+            recursive=True
+        )
         observer.start()
         
         try:
             while True:
-                # 定期检查远程文件删除
                 self.sync_deleted_files()
                 time.sleep(self.config.sync_interval)
-        except KeyboardInterrupt:
-            logger.info("收到停止信号，关闭监控...")
-            # 取消所有待处理的上传计时器
-            with self.upload_queue_lock:
-                for file_path, (_, _, timer) in self.upload_queue.items():
-                    timer.cancel()
-                self.upload_queue.clear()
+        finally:
             observer.stop()
-        observer.join()
-        self.executor.shutdown()
+            observer.join()
+            self.executor.shutdown()
 
 def main():
     if len(sys.argv) < 2:
@@ -486,21 +332,10 @@ def main():
         sys.exit(1)
     
     try:
-        config = Config()
-    except ValueError as e:
-        logger.error(str(e))
-        sys.exit(1)
-    
-    r2_sync = R2Sync(config)
-    command = sys.argv[1]
-    
-    if command == "init":
-        r2_sync.init_sync()
-    elif command == "sync":
-        logger.info(f"启动同步服务，间隔: {config.sync_interval}秒")
-        r2_sync.watch_and_sync()
-    else:
-        logger.error(f"未知命令: {command}")
+        r2_sync = R2Sync(Config())
+        {"init": r2_sync.init_sync, "sync": r2_sync.watch_and_sync}[sys.argv[1]]()
+    except Exception as e:
+        logger.error(f"运行失败: {e}")
         sys.exit(1)
 
 if __name__ == "__main__":