92 lines
2.8 KiB
Python
92 lines
2.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, _format_connect_error(str(exc)), []
|
|
|
|
try:
|
|
ensure_dir_exists(input_dir, samba_conn=samba_conn)
|
|
except Exception as exc: # noqa: BLE001
|
|
return (
|
|
False,
|
|
(f"Ошибка этапа создания входной папки Input: {input_dir}. {exc}"),
|
|
[],
|
|
)
|
|
if not samba_conn.path_exists(input_dir):
|
|
return (
|
|
False,
|
|
(
|
|
"Ошибка этапа создания входной папки Input: "
|
|
f"{input_dir}. Проверьте права на родительский каталог."
|
|
),
|
|
[],
|
|
)
|
|
|
|
ok, error = check_input_dir_access(Path(input_dir), samba_conn=samba_conn)
|
|
if not ok:
|
|
return (
|
|
False,
|
|
(
|
|
"Ошибка этапа проверки доступа к папке Input: "
|
|
f"{error or 'нет прав на чтение/листинг.'}"
|
|
),
|
|
[],
|
|
)
|
|
|
|
try:
|
|
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)
|
|
except Exception as exc: # noqa: BLE001
|
|
return (
|
|
False,
|
|
(f"Ошибка этапа чтения содержимого папки Input: {input_dir}. {exc}"),
|
|
[],
|
|
)
|
|
|
|
return True, "", files
|
|
|
|
|
|
def _format_connect_error(details: str) -> str:
|
|
normalized = details.lower()
|
|
auth_hints = (
|
|
"status_logon_failure",
|
|
"logon failure",
|
|
"authentication",
|
|
"no username or password",
|
|
)
|
|
if any(hint in normalized for hint in auth_hints):
|
|
return f"Ошибка этапа подключения к SMB (аутентификация). {details}"
|
|
return f"Ошибка этапа подключения к SMB. {details}"
|