Delete r2_sync.py

r2_sync.py

@@ -1,531 +0,0 @@
-#!/usr/bin/env python3
-import os
-import sys
-import time
-import requests
-import json
-import hashlib
-import logging
-import threading
-from pathlib import Path
-from concurrent.futures import ThreadPoolExecutor
-from watchdog.observers import Observer
-from watchdog.events import FileSystemEventHandler
-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()
-    # ]
-)
-logger = logging.getLogger(__name__)
-
-class Config:
-    """配置管理类"""
-    def __init__(self):
-        self.data_dir = Path(os.getenv("DATA_DIR", "/app"))
-        self.api_key = os.getenv("API_AUTH_KEY")
-        self.api_endpoint = os.getenv("API_ENDPOINT")
-        self.sync_interval = int(os.getenv("SYNC_INTERVAL", "300"))  # 默认5分钟
-        self.use_hash_check = os.getenv("USE_HASH_CHECK", "true").lower() == "true"
-        self.max_upload_workers = int(os.getenv("MAX_UPLOAD_WORKERS", "500"))
-        self.upload_retry_delay = int(os.getenv("UPLOAD_RETRY_DELAY", "60"))  # 秒
-        
-        # 忽略的文件后缀配置（逗号分隔，如 ".tmp,.log"）
-        ignore_suffixes = os.getenv("IGNORE_SUFFIXES", ".db-journal,.tmp,.log,sqlite3-journal")
-        self.ignore_suffixes = [s.strip().lower() for s in ignore_suffixes.split(",") if s.strip()]
-        
-        # 只同步的文件后缀配置（逗号分隔，如 ".jpg,.png"，优先级高于忽略后缀）
-        only_suffixes = os.getenv("ONLY_SUFFIXES", ".db")
-        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")
-
-class SyncHandler(FileSystemEventHandler):
-    """文件系统事件处理器"""
-    def __init__(self, callback, config):
-        self.callback = callback
-        self.config = config
-    
-    def should_sync(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
-    
-    def on_modified(self, event):
-        if not event.is_directory and self.should_sync(event.src_path):
-            self.callback(event.src_path)
-    
-    def on_created(self, event):
-        if not event.is_directory and self.should_sync(event.src_path):
-            self.callback(event.src_path)
-
-class R2Sync:
-    """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.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
-    
-    def r2_api_request(self, method, path, data=None, max_retries=3):
-        """发送API请求"""
-        url = f"{self.config.api_endpoint}/{path}"
-        headers = {
-            'X-API-Key': self.config.api_key,
-            'Content-Type': 'application/octet-stream'
-        }
-        
-        for attempt in range(max_retries):
-            try:
-                if method == "GET":
-                    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()
-                
-            except requests.exceptions.RequestException as e:
-                if attempt == max_retries - 1:
-                    logger.error(f"API请求最终失败: {e}")
-                    return None
-                wait_time = (attempt + 1) * 2
-                logger.warning(f"API请求失败(尝试 {attempt + 1}/{max_retries}), {wait_time}秒后重试...")
-                time.sleep(wait_time)
-    
-    def get_file_hash(self, file_path):
-        """计算文件哈希值"""
-        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 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()
-                self.cache_valid = True
-            return self.remote_files_cache.copy()
-    
-    def _fetch_remote_files(self):
-        """实际获取远程文件列表"""
-        resp = self.r2_api_request("GET", "list")
-        if not resp:
-            return None
-            
-        try:
-            raw_resp = resp.decode('utf-8').strip()
-            if raw_resp == "[]":
-                return []
-                
-            data = json.loads(raw_resp)
-            
-            # 统一处理为列表格式
-            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
-        except Exception as 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}")
-            return False
-            
-        path = f"download/{key}"
-        logger.debug(f"下载路径: {dest_path} (原始key: {key})")
-        
-        resp = self.r2_api_request("GET", path)
-        if not resp:
-            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)
-        
-        logger.info(f"文件已保存到: {dest_path} (大小: {len(content)}字节)")
-        return True
-    
-    def _upload_file_task(self, file_path, key):
-        """线程池中执行的上传任务"""
-        if not self.should_sync_file(file_path):
-            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
-        except Exception as 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}")
-            return False
-            
-        logger.info(f"删除远程文件: {key}")
-        path = f"delete/{key}"
-        if not self.r2_api_request("DELETE", path):
-            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 _download_and_update_state(self, file_info):
-        """线程任务：下载文件并更新状态"""
-        key = file_info['key']
-        dest_path = self.config.data_dir / key
-        
-        if not self.should_sync_file(dest_path):
-            logger.debug(f"忽略下载文件（不匹配同步规则）: {key}")
-            return
-            
-        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),)
-            with self.cache_lock:  # 使用锁保护file_states更新
-                self.file_states[key] = file_state
-
-    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("获取远程文件列表失败")
-            sys.exit(1)
-            
-        logger.info(f"找到 {len(remote_files)} 个远程文件")
-        if not remote_files:
-            logger.info("远程存储桶为空，无需同步")
-            return
-        
-        logger.info("开始并行同步远程文件...")
-        # 使用线程池并行下载
-        futures = []
-        for file_info in remote_files:
-            futures.append(self.executor.submit(self._download_and_update_state, file_info))
-            
-        # 等待所有任务完成
-        for future in futures:
-            try:
-                future.result()  # 捕获任何异常
-            except Exception as e:
-                logger.error(f"文件下载任务出错: {e}")
-    
-    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
-            
-            with self.upload_queue_lock:
-                # 如果文件已在队列中，更新修改时间并重置计时器
-                if file_path in self.upload_queue:
-                    _, _, timer = self.upload_queue[file_path]
-                    timer.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}秒后检查)")
-                
-        except Exception as e:
-            logger.error(f"处理文件变化出错: {e}")
-    def is_file_modified(self, file_path, last_known_state):
-            """检查文件是否修改"""
-            try:
-                stat = os.stat(file_path)
-                current_size = stat.st_size
-                
-                # 只比较文件大小
-                if last_known_state[0] == current_size:  # 比较size
-                    logger.info(f"文件大小没有变化: {file_path} (大小: {current_size})")
-                    # 新增：当hash不存在时，计算并更新hash
-                    if not self.config.use_hash_check:
-                        logger.info(f"不使用hash检查: {file_path} 视为未修改")
-                        return False  # 不使用hash检查，直接返回False
-                    logger.info(f"文件大小相同，开始检查hash: {file_path}")
-                    current_hash = self.get_file_hash(file_path)
-                    if len(last_known_state) <= 2:  # hash不存在
-                        # 获取文件key用于更新状态
-                        rel_path = str(Path(file_path).relative_to(self.config.data_dir)).replace('\\', '/')
-                        # 更新状态 (size, mtime, hash)
-                        self.file_states[rel_path] = (current_size, stat.st_mtime, current_hash)
-                        return True  # 返回True触发更新
-                    return current_hash != last_known_state[2]  # 比较hash
-                return True  # 大小不同，认为已修改
-            except Exception as e:
-                logger.error(f"检查文件修改状态出错: {e}")
-                return False
-    def _process_upload_queue(self, file_path):
-        """处理上传队列中的文件"""
-        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]  # 从队列移除
-            
-        try:
-            # 检查文件是否还存在
-            if not os.path.exists(file_path):
-                logger.info(f"文件已被删除，取消上传: {key}")
-                return
-                
-            # 获取当前文件状态
-            current_stat = os.stat(file_path)
-            current_mtime = current_stat.st_mtime
-            
-            # 比较修改时间，确认文件是否又有新变化
-            if current_mtime != original_mtime:
-                logger.info(f"文件 {key} 在等待期间又有新变化，重新加入队列")
-                self.handle_file_change(file_path)  # 重新加入队列
-                return
-            
-            if not self.is_file_modified(file_path, self.file_states.get(key, (0, 0))):
-                logger.info(f"文件 {key} 没有新变化，跳过上传")
-                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}")
-        except Exception as 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
-        
-        # 设置文件监控（根据同步规则）
-        event_handler = SyncHandler(self.handle_file_change, self.config)
-        observer = Observer()
-        observer.schedule(event_handler, 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()
-            observer.stop()
-        observer.join()
-        self.executor.shutdown()
-
-def main():
-    if len(sys.argv) < 2:
-        logger.info("Usage: python r2_sync.py [init|sync]")
-        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}")
-        sys.exit(1)
-
-if __name__ == "__main__":
-    main()