diff --git a/app/pipeline/file_manager.py b/app/pipeline/file_manager.py index d74c01a..f5fa34b 100644 --- a/app/pipeline/file_manager.py +++ b/app/pipeline/file_manager.py @@ -38,6 +38,7 @@ class RemoteFile: class InputFileScanResult: valid_files: list[RemoteFile] invalid_files: list[RemoteFile] + skipped_files: list[RemoteFile] class SMBClientConnection: @@ -285,20 +286,32 @@ def next_run_number(out_day_dir: str, samba_conn) -> int: return max(numbers, default=0) + 1 -def scan_input_files(input_dir: str, samba_conn) -> InputFileScanResult: +def scan_input_files( + input_dir: str, samba_conn, processed_dir: str | None = None +) -> InputFileScanResult: valid_files: list[RemoteFile] = [] invalid_files: list[RemoteFile] = [] + skipped_files: list[RemoteFile] = [] for file_name in _list_samba_files(input_dir, samba_conn): if not file_name.lower().endswith(".xml"): continue remote_path = join_path(input_dir, file_name) + if processed_dir and _is_already_processed( + file_name, processed_dir, samba_conn + ): + skipped_files.append(RemoteFile(path=remote_path, name=file_name)) + continue if validate_file_name(file_name): valid_files.append(RemoteFile(path=remote_path, name=file_name)) else: invalid_files.append(RemoteFile(path=remote_path, name=file_name)) - return InputFileScanResult(valid_files=valid_files, invalid_files=invalid_files) + return InputFileScanResult( + valid_files=valid_files, + invalid_files=invalid_files, + skipped_files=skipped_files, + ) def move_file_to_error( @@ -393,6 +406,11 @@ def _list_samba_files(input_dir: str, samba_conn) -> list[str]: return items +def _is_already_processed(file_name: str, processed_dir: str, samba_conn) -> bool: + with samba_conn: + return samba_conn.path_exists(join_path(processed_dir, file_name)) + + def _next_available_samba_target(dir_path: str, file_name: str, samba_conn) -> str: with samba_conn: if not samba_conn.path_exists(dir_path): diff --git a/app/pipeline/service.py b/app/pipeline/service.py index 9122786..a58030f 100644 --- a/app/pipeline/service.py +++ b/app/pipeline/service.py @@ -120,9 +120,15 @@ def run_batch( log_lines: list[str] = [] log_lines.append(f"{_now()} | preflight access checks: ok") - scan_result = scan_input_files(input_dir=input_dir, samba_conn=samba_conn) + scan_result = scan_input_files( + input_dir=input_dir, + samba_conn=samba_conn, + processed_dir=run_context.processed_dir, + ) log_lines.append( - f"{_now()} | scan files: valid={len(scan_result.valid_files)} invalid={len(scan_result.invalid_files)} " + f"{_now()} | scan files: valid={len(scan_result.valid_files)} " + f"invalid={len(scan_result.invalid_files)} " + f"skipped={len(scan_result.skipped_files)}" ) report_file_name = build_report_file_name( @@ -208,7 +214,10 @@ def run_batch( ) messages.append( _build_result_message( - status=status, processed=processed, partial=partial, errors=errors + status=status, + processed=processed, + partial=partial, + errors=errors, ) ) duration_seconds = round(time.perf_counter() - started_at, 2) @@ -254,6 +263,9 @@ def _process_files( log_lines: list[str], stage_timings: dict[str, float], ) -> None: + for remote_file in scan_result.skipped_files: + log_lines.append(f"{_now()} | skip already processed: {remote_file.path}") + for remote_file in scan_result.invalid_files: reason = "Некорректное имя XML-файла" moved_to = "" diff --git a/tests/unit/test_file_manager.py b/tests/unit/test_file_manager.py index 238fb1c..b3408fa 100644 --- a/tests/unit/test_file_manager.py +++ b/tests/unit/test_file_manager.py @@ -85,6 +85,26 @@ def test_scan_input_files_splits_valid_and_invalid(tmp_path: Path) -> None: result = scan_input_files(in_dir, samba) assert len(result.valid_files) == 1 assert len(result.invalid_files) == 1 + assert len(result.skipped_files) == 0 + + +def test_scan_input_files_skips_already_processed(tmp_path: Path) -> None: + samba = FakeSamba(tmp_path) + in_dir = "Exchange\\SMBDEMO\\test" + processed_dir = "Exchange\\SMBDEMO\\Processed" + local_in_dir = tmp_path / "Exchange" / "SMBDEMO" / "test" + local_processed_dir = tmp_path / "Exchange" / "SMBDEMO" / "Processed" + local_in_dir.mkdir(parents=True) + local_processed_dir.mkdir(parents=True) + + file_name = "SKO115FZ_01_123456789_20260616_X00001.xml" + (local_in_dir / file_name).write_text("ok", encoding="utf-8") + (local_processed_dir / file_name).write_text("already done", encoding="utf-8") + + result = scan_input_files(in_dir, samba, processed_dir=processed_dir) + assert len(result.valid_files) == 0 + assert len(result.invalid_files) == 0 + assert [item.name for item in result.skipped_files] == [file_name] def test_next_run_number_reads_existing_dirs(tmp_path: Path) -> None: