61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from app.pipeline.file_manager import (
|
||
check_input_dir_access,
|
||
connect_samba,
|
||
ensure_dir_exists,
|
||
)
|
||
|
||
|
||
def build_user_params(samba_user: str, samba_password: str) -> dict[str, str]:
|
||
return {
|
||
"user_name": samba_user,
|
||
"password": samba_password,
|
||
}
|
||
|
||
|
||
def check_access_and_list_files(
|
||
smb_base_path: str,
|
||
input_dir: str,
|
||
samba_user: str,
|
||
samba_password: str,
|
||
) -> tuple[bool, str, list[str]]:
|
||
if not samba_user:
|
||
return False, "Введите логин SAMBA.", []
|
||
|
||
user_params = build_user_params(samba_user, samba_password)
|
||
try:
|
||
samba_conn = connect_samba(smb_base_path=smb_base_path, user_params=user_params)
|
||
except RuntimeError as exc:
|
||
return False, str(exc), []
|
||
|
||
try:
|
||
ensure_dir_exists(input_dir, samba_conn=samba_conn)
|
||
except Exception as exc: # noqa: BLE001
|
||
return False, f"Не удалось создать входную папку: {input_dir}. {exc}", []
|
||
if not samba_conn.path_exists(input_dir):
|
||
return (
|
||
False,
|
||
(
|
||
"Не удалось создать входную папку: "
|
||
f"{input_dir}. Проверьте права на родительский каталог."
|
||
),
|
||
[],
|
||
)
|
||
|
||
ok, error = check_input_dir_access(Path(input_dir), samba_conn=samba_conn)
|
||
if not ok:
|
||
return False, error or "Ошибка доступа к входной папке.", []
|
||
|
||
with samba_conn:
|
||
entries = sorted(samba_conn.listdir(input_dir))
|
||
files: list[str] = []
|
||
for name in entries:
|
||
full_path = input_dir.rstrip("/\\") + "\\" + name
|
||
if samba_conn.is_file(full_path):
|
||
files.append(name)
|
||
|
||
return True, "", files
|