43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from app.pipeline.file_manager import check_input_dir_access, connect_samba
|
|
|
|
|
|
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), []
|
|
|
|
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
|