stage 3
This commit is contained in:
parent
6f6ed4afc0
commit
d454cf1577
@ -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):
|
||||
|
||||
@ -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 = ""
|
||||
|
||||
@ -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:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user