diff --git a/client-file-collector/client_file_collector.py b/client-file-collector/client_file_collector.py index 6273d52..9115bcb 100644 --- a/client-file-collector/client_file_collector.py +++ b/client-file-collector/client_file_collector.py @@ -140,6 +140,8 @@ class CollectorApp: self.active_rows = [] self.upload_state = {} self.notice_ts = {} + self.processing_lock = threading.Lock() + self.processing_files = set() self.http_session = self._create_http_session() self.base_url_var = StringVar() @@ -421,6 +423,8 @@ class CollectorApp: # 每次手动/自动启动都重置一次状态,避免停启后误判“已同步”。 self.upload_state = {} self.notice_ts = {} + with self.processing_lock: + self.processing_files.clear() self.stop_event = threading.Event() self.running = True self.worker_thread = threading.Thread(target=self._worker_loop, args=(self.stop_event,), daemon=True) @@ -521,20 +525,28 @@ class CollectorApp: for file_path in files: if stop_event.is_set(): return - signature = self._file_signature(file_path) - if signature is None: - continue - state_key = "%d|%s" % (row_index, str(file_path).lower()) - if self.upload_state.get(state_key) == signature: + if not self._try_acquire_processing(state_key): continue - if self._upload_file(row, file_path): - if self._backup_and_remove_file(row, file_path): - # 文件已从源目录移走,立即清理状态,允许后续同名新文件再次同步。 - self.upload_state.pop(state_key, None) - else: - self.upload_state.pop(state_key, None) + try: + signature = self._file_signature(file_path) + if signature is None: + continue + + # 已上传但上次备份失败,则仅重试备份,避免重复上传造成服务器/备份放大。 + if self.upload_state.get(state_key) == signature: + if self._backup_and_remove_file(row, file_path): + self.upload_state.pop(state_key, None) + continue + + if self._upload_file(row, file_path): + self.upload_state[state_key] = signature + if self._backup_and_remove_file(row, file_path): + # 文件已从源目录移走,立即清理状态,允许后续同名新文件再次同步。 + self.upload_state.pop(state_key, None) + finally: + self._release_processing(state_key) def _cleanup_row_upload_state(self, row_index, files): prefix = "%d|" % row_index @@ -543,13 +555,24 @@ class CollectorApp: current_keys.add("%d|%s" % (row_index, str(file_path).lower())) stale_keys = [] - for key in self.upload_state.keys(): + for key in list(self.upload_state.keys()): if key.startswith(prefix) and key not in current_keys: stale_keys.append(key) for key in stale_keys: self.upload_state.pop(key, None) + def _try_acquire_processing(self, state_key): + with self.processing_lock: + if state_key in self.processing_files: + return False + self.processing_files.add(state_key) + return True + + def _release_processing(self, state_key): + with self.processing_lock: + self.processing_files.discard(state_key) + def _file_signature(self, file_path): try: stat = file_path.stat() @@ -585,19 +608,38 @@ class CollectorApp: self.log("第%d行创建备份目录失败: %s, 原因: %s" % (row_index, backup_dir, e)) return False - target = backup_dir / file_path.name - if target.exists(): - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - target = backup_dir / ("%s_%s%s" % (file_path.stem, timestamp, file_path.suffix)) + if not file_path.exists(): + self.log("第%d行源文件已不存在,视为已备份完成: %s" % (row_index, file_path.name)) + return True + + target = self._build_unique_backup_target(backup_dir, file_path) try: shutil.move(str(file_path), str(target)) self.log("第%d行源文件已转移至备份: %s -> %s" % (row_index, file_path.name, target)) return True + except FileNotFoundError: + self.log("第%d行源文件在备份时已被处理,忽略重复备份: %s" % (row_index, file_path.name)) + return True except Exception as e: self.log("第%d行源文件备份转移失败: %s, 原因: %s" % (row_index, file_path.name, e)) return False + def _build_unique_backup_target(self, backup_dir, file_path): + target = backup_dir / file_path.name + if not target.exists(): + return target + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + base_name = file_path.stem + suffix = file_path.suffix + candidate = backup_dir / ("%s_%s%s" % (base_name, timestamp, suffix)) + index = 1 + while candidate.exists(): + candidate = backup_dir / ("%s_%s_%d%s" % (base_name, timestamp, index, suffix)) + index += 1 + return candidate + def _upload_file(self, row, file_path): url = self._base_url() + "/collector/client/upload" data = {