采集程序在运行一段时间后备份的时候翻倍复制文件

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 = {