feat(smb): запись прямо в Input с перезаписью, как в старом АРМ (#26)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-09 16:30:31 +04:00
parent 7ec3dd49fd
commit 6870ca799a
4 changed files with 62 additions and 155 deletions

View File

@ -31,7 +31,6 @@ def create_hotfolder(settings: Settings) -> HotFolder:
smb.SMB_PATH,
settings.SCAN_INPUT_DIR,
settings.SCAN_OUTPUT_DIR,
settings.SCAN_TMP_DIR,
login=smb.SMB_LOGIN,
password=smb.SMB_PASSWORD,
settle_seconds=settings.SCAN_SETTLE_SECONDS,

View File

@ -4,10 +4,10 @@ r"""Горячая папка на SMB-шаре платформы (`SCAN_HOTFOL
в lifespan приложения (`app.py`); логин/пароль, если заданы, передаются ещё и в каждый вызов `smbclient`, чтобы
сессия поднималась заново по требованию (после сбоя регистрации при старте или сброса соединения).
Пути — UNC: временный файл `\\<host>\<SMB_PATH>\<tmp_dir>\.<имя>.part` переименовывается (rename без замены)
во входящий `\\<host>\<SMB_PATH>\<input_dir>\<имя>` (одна шара — rename атомарен, OQ-79); ответ — все `*.json`
каталога `\\<host>\<SMB_PATH>\<output_dir>\<job_id>\` (OQ-78), готовность — по стабилизации набора
(`SettleTracker`). Очистка (D-2) удаляет ответы, входящий файл и временные файлы при `SCAN_CLEANUP=true`.
Пути — UNC: входящий файл пишется сразу в `\\<host>\<SMB_PATH>\<input_dir>\<имя>` (`wb`, существующий перезаписывается,
как в старом АРМ — решение владельца 09.10.2026, change `hotfolder-direct-write`); ответ — все `*.json` каталога
`\\<host>\<SMB_PATH>\<output_dir>\<job_id>\` (OQ-78), готовность — по стабилизации набора (`SettleTracker`).
Очистка (D-2) удаляет ответы, входящий файл и его `.done` при `SCAN_CLEANUP=true`.
Ошибки smbprotocol — `SMBOSError` (подкласс `OSError`, но НЕ `FileNotFoundError`): «нет файла/каталога» распознаётся
по `errno == ENOENT`.
"""
@ -66,9 +66,9 @@ def build_unc_root(host: str, path: str) -> str:
class SmbHotFolder:
"""Горячая папка ИС Сканирование на SMB-шаре. Методы синхронные (блокирующий SMB), семантика как у
`LocalHotFolder`: те же правила имён, атомарная запись `Tmp\\.<name>.part` + rename без замены в `Input`,
`Tmp` и `Input` создаются без родителей, корень не создаётся, готовность набора ответа — по стабилизации
(`SettleTracker`). Очистка (D-2) под флагом `SCAN_CLEANUP`."""
`LocalHotFolder`: те же правила имён, запись сразу в `Input` с перезаписью, `Input` создаётся без родителей, корень
не создаётся, готовность набора ответа — по стабилизации (`SettleTracker`). Очистка (D-2) под флагом
`SCAN_CLEANUP`."""
def __init__( # noqa: PLR0913 — адрес и учётные данные D-1 + имена каталогов и готовность, как у LocalHotFolder
self,
@ -76,7 +76,6 @@ class SmbHotFolder:
path: str,
input_dir: str = "Input",
output_dir: str = "Output",
tmp_dir: str = "Tmp",
*,
login: str = "",
password: str = "",
@ -87,7 +86,6 @@ class SmbHotFolder:
self.root: str = build_unc_root(host, path)
self.input_dir: str = ntpath.join(self.root, input_dir)
self.output_dir: str = ntpath.join(self.root, output_dir)
self.tmp_dir: str = ntpath.join(self.root, tmp_dir)
self._credentials: dict[str, Any] = {"username": login, "password": password} if login else {}
self._settle = SettleTracker(settle_seconds, clock)
self.cleanup_enabled = cleanup
@ -102,29 +100,20 @@ class SmbHotFolder:
return ntpath.join(self.answer_dir(job_id), f"{job_id}.json")
def put(self, name: str, data: bytes) -> None:
"""Записать `.<name>.part` в служебный каталог, затем `smbclient.rename` в `<input>\\<name>` без замены.
Имя проверяется до любого обращения к шаре. Назначение уже есть (повтор задания) — это не ошибка: файл
считается доставленным и не заменяется. В этом случае, как и при любой другой ошибке записи или переименования,
`.part` снимается только при включённой очистке (`SCAN_CLEANUP`, D-2), иначе остаётся в `Tmp`; ошибка снятия
исходную не маскирует. При успехе `.part` забирает сам rename. Нет корня — `SMBOSError` (ENOENT) от первого
`mkdir`.
"""Записать `<input>\\<name>` сразу, открытием `wb` (как в старом АРМ, решение владельца 09.10.2026): временного
файла и rename нет, существующий файл перезаписывается. `smbclient.open_file` открывает файл монопольно — пока
запись идёт, ИС его не прочитает (ASSUMPTION(OQ-79)). Имя проверяется до любого обращения к шаре. При ошибке
записи недописанный файл снимается только при включённой очистке (`SCAN_CLEANUP`, R-3); ошибка снятия исходную
не маскирует. Нет корня — `SMBOSError` (ENOENT) от `mkdir`.
"""
validate_name(name)
self._ensure_dir(self.tmp_dir)
self._ensure_dir(self.input_dir)
part = ntpath.join(self.tmp_dir, f".{name}.part")
target = ntpath.join(self.input_dir, name)
try:
with smbclient.open_file(part, mode="wb", **self._credentials) as fd:
with smbclient.open_file(target, mode="wb", **self._credentials) as fd:
fd.write(data)
smbclient.rename(part, ntpath.join(self.input_dir, name), **self._credentials)
except OSError as err:
self._drop_part(part)
ntstatus = getattr(err, "ntstatus", None)
if _has_errno(err, errno.EEXIST, FileExistsError) or ntstatus == NtStatus.STATUS_OBJECT_NAME_COLLISION:
return
raise
except BaseException:
self._drop_part(part)
self._drop_partial(target)
raise
def poll(self, job_id: str) -> list[AnswerFile] | None:
@ -166,14 +155,14 @@ class SmbHotFolder:
"""Число заданий с ненастоявшимся набором (для тестов и диагностики)."""
return len(self._settle)
def _drop_part(self, path: str) -> None:
"""Удалить временный файл при ошибке, только если очистка включена (D-2)."""
def _drop_partial(self, path: str) -> None:
"""Удалить недописанный входящий файл при ошибке записи, только если очистка включена (D-2, R-3)."""
if self.cleanup_enabled:
with contextlib.suppress(Exception):
smbclient.remove(path, **self._credentials)
def cleanup(self, job_id: str, name: str) -> bool:
"""Удалить следы задания: ответ из `output`, входящий файл, временные файлы. Возвращает `False`, если
"""Удалить следы задания: ответ из `output`, входящий файл и его `.done`. Возвращает `False`, если
очистка выключена; `True` при успешной очистке. Ошибки удаления (кроме ENOENT) поднимаются."""
if not self.cleanup_enabled:
return False
@ -194,7 +183,6 @@ class SmbHotFolder:
for path in (
ntpath.join(self.input_dir, name),
ntpath.join(self.input_dir, f"{name}.done"),
ntpath.join(self.tmp_dir, f".{name}.part"),
):
try:
smbclient.remove(path, **self._credentials)

View File

@ -33,7 +33,6 @@ def _smb_settings() -> Settings:
update={
"SCAN_HOTFOLDER_MODE": "smb",
"SCAN_OUTPUT_DIR": "Out",
"SCAN_TMP_DIR": "T",
"SCAN_SETTLE_SECONDS": 3,
"OPENBAO": settings.OPENBAO.model_copy(update={"SMB": smb}),
}
@ -66,7 +65,7 @@ def test_smb_mode_builds_smb_hotfolder() -> None:
assert hf.root == "\\\\smb.example.local\\share\\scan"
assert hf.input_dir == "\\\\smb.example.local\\share\\scan\\Input"
assert hf.output_dir == "\\\\smb.example.local\\share\\scan\\Out"
assert hf.tmp_dir == "\\\\smb.example.local\\share\\scan\\T"
assert not hasattr(hf, "tmp_dir") # у шлюза служебного каталога нет (change hotfolder-direct-write)
assert hf.settle_seconds == 3

View File

@ -21,7 +21,6 @@ HOST = "smb.example.local"
ROOT = "\\\\smb.example.local\\share\\scan"
INPUT = ROOT + "\\Input"
OUTPUT = ROOT + "\\Output"
TMP = ROOT + "\\Tmp"
class FakeClock:
@ -178,7 +177,6 @@ class TestPaths:
assert hf.root == ROOT
assert hf.input_dir == INPUT
assert hf.output_dir == OUTPUT
assert hf.tmp_dir == TMP
assert hf.answer_dir(job_id) == f"{OUTPUT}\\{job_id}"
assert hf.answer_path(job_id) == f"{OUTPUT}\\{job_id}\\{job_id}.json"
@ -194,10 +192,9 @@ class TestPaths:
assert SmbHotFolder(HOST, "share").root == "\\\\smb.example.local\\share"
def test_custom_dir_names(self) -> None:
hf = SmbHotFolder(HOST, "share/scan", input_dir="Входящие", output_dir="Исходящие", tmp_dir="Служебный")
hf = SmbHotFolder(HOST, "share/scan", input_dir="Входящие", output_dir="Исходящие")
assert hf.input_dir == ROOT + "\\Входящие"
assert hf.output_dir == ROOT + "\\Исходящие"
assert hf.tmp_dir == ROOT + "\\Служебный"
assert hf.answer_path("abc") == ROOT + "\\Исходящие\\abc\\abc.json"
@pytest.mark.parametrize(("host", "path"), [(HOST, ""), (HOST, "/"), ("", "share/scan"), ("", "\\\\srv")])
@ -226,63 +223,13 @@ class TestPaths:
class TestPut:
def test_writes_part_in_tmp_then_renames_into_input(self, fake: FakeSmbClient, share_root: Path) -> None:
def test_writes_directly_into_input(self, fake: FakeSmbClient, share_root: Path) -> None:
name = inbox_name(_job_id(), "aep_5.PDF")
data = bytes(range(256)) * 10
_hotfolder().put(name, data)
part = f"{TMP}\\.{name}.part"
assert fake.calls == [
("mkdir", (TMP,), {}),
("mkdir", (INPUT,), {}),
("open_file", (part, "wb"), {}),
("rename", (part, f"{INPUT}\\{name}"), {}),
]
assert fake.calls == [("mkdir", (INPUT,), {}), ("open_file", (f"{INPUT}\\{name}", "wb"), {})]
assert (share_root / "Input" / name).read_bytes() == data
assert [p.name for p in (share_root / "Input").iterdir()] == [name]
assert list((share_root / "Tmp").iterdir()) == []
def test_input_has_no_part_during_write(self, fake: FakeSmbClient, share_root: Path) -> None:
"""OQ-79: пока пишется временный файл, во входящем каталоге его нет — он в `Tmp`."""
name = "a.pdf"
seen: dict[str, list[str]] = {}
real_rename = fake.rename
def spy(src: str, dst: str, **kwargs: Any) -> None:
seen["input"] = sorted(p.name for p in (share_root / "Input").iterdir())
seen["tmp"] = sorted(p.name for p in (share_root / "Tmp").iterdir())
real_rename(src, dst, **kwargs)
fake.rename = spy # type: ignore[method-assign]
_hotfolder().put(name, b"data")
assert seen == {"input": [], "tmp": [".a.pdf.part"]}
assert [p.name for p in (share_root / "Input").iterdir()] == [name]
def test_existing_tmp_dir_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
(share_root / "Tmp").mkdir()
_hotfolder().put("a.pdf", b"1")
assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
def test_tmp_mkdir_denied_but_tmp_exists_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
(share_root / "Tmp").mkdir()
(share_root / "Input").mkdir()
fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP)
_hotfolder().put("a.pdf", b"1")
assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
assert fake.calls[:4] == [
("mkdir", (TMP,), {}),
("isdir", (TMP,), {}),
("mkdir", (INPUT,), {}),
("isdir", (INPUT,), {}),
]
def test_tmp_mkdir_denied_and_no_tmp_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
(share_root / "Input").mkdir()
denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP)
fake.fail["mkdir"] = denied
with pytest.raises(SMBOSError) as excinfo:
_hotfolder().put("a.pdf", b"1")
assert excinfo.value is denied
assert fake.funcs() == ["mkdir", "isdir"]
assert sorted(p.name for p in share_root.iterdir()) == ["Input"] # Tmp не создаётся
def test_existing_input_dir_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
(share_root / "Input").mkdir()
@ -298,41 +245,29 @@ class TestPut:
def test_mkdir_denied_but_input_exists_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
"""Урезанные ACL: нельзя создавать подкаталоги в корне, но `Input` уже есть и в него писать можно."""
(share_root / "Tmp").mkdir()
(share_root / "Input").mkdir()
fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
_hotfolder().put("a.pdf", b"1")
assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
assert fake.funcs() == ["mkdir", "isdir", "mkdir", "isdir", "open_file", "rename"]
assert fake.calls[3] == ("isdir", (INPUT,), {})
assert fake.funcs() == ["mkdir", "isdir", "open_file"]
def test_mkdir_denied_and_no_input_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
(share_root / "Tmp").mkdir()
denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
fake.fail["mkdir"] = denied
with pytest.raises(SMBOSError) as excinfo:
_hotfolder().put("a.pdf", b"1")
assert excinfo.value is denied
assert fake.funcs() == ["mkdir", "isdir", "mkdir", "isdir"]
assert fake.funcs() == ["mkdir", "isdir"]
def test_mkdir_denied_and_isdir_fails_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP)
denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
fake.fail["mkdir"] = denied
fake.fail["isdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, TMP)
fake.fail["isdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
with pytest.raises(SMBOSError) as excinfo:
_hotfolder().put("a.pdf", b"1")
assert excinfo.value is denied
assert fake.funcs() == ["mkdir", "isdir"]
@pytest.mark.parametrize("plain", [False, True], ids=["SMBOSError", "FileExistsError"])
def test_no_replace_on_existing_file(self, fake: FakeSmbClient, share_root: Path, plain: bool) -> None:
fake.plain_collision = plain
hf = _hotfolder()
hf.put("a.pdf", b"old")
hf.put("a.pdf", b"new")
assert (share_root / "Input" / "a.pdf").read_bytes() == b"old"
assert (share_root / "Tmp" / ".a.pdf.part").read_bytes() == b"new"
def test_empty_data_is_written(self, fake: FakeSmbClient, share_root: Path) -> None:
_hotfolder().put("empty.pdf", b"")
assert (share_root / "Input" / "empty.pdf").read_bytes() == b""
@ -371,58 +306,44 @@ class TestPut:
_hotfolder().put(name, b"x")
assert fake.calls == []
def test_cleans_part_file_on_rename_failure(self, fake: FakeSmbClient, share_root: Path) -> None:
fake.fail["rename"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
with pytest.raises(OSError):
_hotfolder(cleanup=True).put("a.pdf", b"data")
part = f"{TMP}\\.a.pdf.part"
assert fake.calls[-1] == ("remove", (part,), {})
assert list((share_root / "Input").iterdir()) == []
assert list((share_root / "Tmp").iterdir()) == []
def test_write_failure_keeps_partial_file_when_cleanup_disabled(
self, fake: FakeSmbClient, share_root: Path
) -> None:
fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
with pytest.raises(SMBOSError):
_hotfolder().put("a.pdf", b"data")
assert "remove" not in fake.funcs()
def test_cleanup_error_does_not_mask_original(self, fake: FakeSmbClient, share_root: Path) -> None:
fake.fail["rename"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
fake.fail["remove"] = SMBException("соединение потеряно")
def test_write_failure_removes_partial_file_when_cleanup_enabled(
self, fake: FakeSmbClient, share_root: Path
) -> None:
fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
with pytest.raises(SMBOSError):
_hotfolder(cleanup=True).put("a.pdf", b"data")
assert fake.funcs() == ["mkdir", "mkdir", "open_file", "rename", "remove"]
assert fake.calls[-1] == ("remove", (f"{INPUT}\\a.pdf",), {})
def test_cleanup_error_does_not_mask_original(self, fake: FakeSmbClient, share_root: Path) -> None:
original = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
fake.fail["open_file"] = original
fake.fail["remove"] = SMBException("соединение потеряно")
with pytest.raises(SMBOSError) as excinfo:
_hotfolder(cleanup=True).put("a.pdf", b"data")
assert excinfo.value is original
assert fake.funcs() == ["mkdir", "open_file", "remove"]
def test_passes_credentials_to_every_call(self, fake: FakeSmbClient, share_root: Path) -> None:
_hotfolder(login="user", password="secret").put("a.pdf", b"1")
assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 4
assert [kwargs for _, _, kwargs in fake.calls] == [{"username": "user", "password": "secret"}] * 2
class TestPutWithoutReplace:
def test_uses_rename_not_replace(self, fake: FakeSmbClient, share_root: Path) -> None:
class TestPutOverwrite:
def test_existing_file_is_overwritten(self, fake: FakeSmbClient, share_root: Path) -> None:
"""Как в старом АРМ (решение владельца 09.10.2026): повтор задания перезаписывает входящий файл."""
hf = _hotfolder()
hf.put("job.pdf", b"x")
names = [call[0] for call in fake.calls]
assert "rename" in names and "replace" not in names
assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"x"
assert not fake._local(TMP + "\\.job.pdf.part").exists()
@pytest.mark.parametrize("plain", [False, True], ids=["SMBOSError", "FileExistsError"])
def test_existing_destination_kept_and_part_left(self, fake: FakeSmbClient, share_root: Path, plain: bool) -> None:
fake.plain_collision = plain
hf = _hotfolder()
hf.put("job.pdf", b"first")
hf.put("job.pdf", b"first, longer")
hf.put("job.pdf", b"second")
assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"first"
assert fake._local(TMP + "\\.job.pdf.part").read_bytes() == b"second"
@pytest.mark.parametrize("plain", [False, True], ids=["SMBOSError", "FileExistsError"])
def test_existing_destination_drops_part_when_cleanup_enabled(
self, fake: FakeSmbClient, share_root: Path, plain: bool
) -> None:
"""Назначение есть (повтор задания) и очистка включена: файл не заменён, `.part` снят."""
fake.plain_collision = plain
hf = _hotfolder(cleanup=True)
hf.put("job.pdf", b"first")
fake.calls.clear()
hf.put("job.pdf", b"second") # исключения нет: назначение есть — файл считается доставленным
assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"first"
assert not fake._local(TMP + "\\.job.pdf.part").exists()
assert fake.calls[-1] == ("remove", (f"{TMP}\\.job.pdf.part",), {})
assert fake._local(INPUT + "\\job.pdf").read_bytes() == b"second"
assert "rename" not in fake.funcs() and "replace" not in fake.funcs()
class TestCleanup:
@ -481,7 +402,7 @@ class TestCleanupDetails:
hf.cleanup(_job_id(), name)
assert fake.calls == []
def test_orphan_part_and_done_marker_are_removed(self, fake: FakeSmbClient) -> None:
def test_done_marker_is_removed_and_tmp_is_not_touched(self, fake: FakeSmbClient) -> None:
hf = self._hf(fake)
job_id = _job_id()
name = f"{job_id}.pdf"
@ -490,14 +411,14 @@ class TestCleanupDetails:
(root / folder).mkdir()
(root / "Input" / name).write_bytes(b"pdf")
(root / "Input" / f"{name}.done").write_bytes(b"") # маркер ИС
(root / "Tmp" / f".{name}.part").write_bytes(b"orphan") # `.part` осиротевшей записи
(root / "Tmp" / f".{name}.part").write_bytes(b"legacy") # остаток прежних версий шлюза — не наш
other = root / "Input" / f"{_job_id()}.pdf"
other.write_bytes(b"other job") # чужое задание не трогаем
assert hf.cleanup(job_id, name) is True
assert sorted(p.name for p in (root / "Input").iterdir()) == [other.name]
assert list((root / "Tmp").iterdir()) == []
assert (root / "Tmp" / f".{name}.part").exists()
removed = [args[0] for func, args, _ in fake.calls if func == "remove"]
assert removed == [f"{INPUT}\\{name}", f"{INPUT}\\{name}.done", f"{TMP}\\.{name}.part"]
assert removed == [f"{INPUT}\\{name}", f"{INPUT}\\{name}.done"]
def test_scandir_access_denied_propagates_and_nothing_else_is_touched(self, fake: FakeSmbClient) -> None:
hf = self._hf(fake)
@ -523,7 +444,7 @@ class TestCleanupDetails:
assert excinfo.value.ntstatus == NtStatus.STATUS_DIRECTORY_NOT_EMPTY
assert not (answer / f"{job_id}.json").exists() # файлы сняты, каталог остался вместе с подкаталогом
assert (answer / "nested").is_dir()
assert fake.funcs()[-1] == "rmdir" # дальше (входящий файл, `.done`, `.part`) очистка не пошла
assert fake.funcs()[-1] == "rmdir" # дальше (входящий файл, `.done`) очистка не пошла
def test_passes_credentials_to_every_call(self, fake: FakeSmbClient) -> None:
hf = self._hf(fake, login="user", password="secret")
@ -536,13 +457,13 @@ class TestCleanupDetails:
(answer / f"{job_id}_ofr.json").write_bytes(b"{}")
fake.calls.clear()
assert hf.cleanup(job_id, name) is True
assert fake.funcs() == ["scandir", "remove", "remove", "rmdir", "remove", "remove", "remove"]
assert fake.funcs() == ["scandir", "remove", "remove", "rmdir", "remove", "remove"]
assert all(kwargs == {"username": "user", "password": "secret"} for _, _, kwargs in fake.calls)
def test_missing_answer_dir_and_files_are_fine(self, fake: FakeSmbClient) -> None:
hf = self._hf(fake)
assert hf.cleanup(_job_id(), "nothing.pdf") is True
assert fake.funcs() == ["scandir", "remove", "remove", "remove"]
assert fake.funcs() == ["scandir", "remove", "remove"]
def test_answer_dir_is_a_file_is_fine(self, fake: FakeSmbClient) -> None:
hf = self._hf(fake)