RSHB/SFM
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: RSHB:636acd1087c7574a2397e36fc6aef340a61db261
Branches Tags
RSHB:parser
..
compare: RSHB:2d31aa1f66040af02b612bc9caf4172883dee6c2
Branches Tags
RSHB:parser

No commits in common. "636acd1087c7574a2397e36fc6aef340a61db261" and "2d31aa1f66040af02b612bc9caf4172883dee6c2" have entirely different histories.
636acd1087 ... 2d31aa1f66

29 changed files with 105 additions and 3952 deletions
Show Stats Expand all files Collapse all files

17
.pre-commit-config.yaml
View File

@ -1,17 +0,0 @@
repos:
- repo: https://github.com/pycqa/isort
rev: 5.13.2
hooks:
- id: isort
args: ["--profile", "black", "--line-length", "88"]
- repo: https://github.com/psf/black
rev: 24.4.2
hooks:
- id: black
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.12.7
hooks:
- id: ruff
- id: ruff-format

76
README.md
View File

@ -1,50 +1,39 @@
# SFM Streamlit
Production-ready Streamlit приложение для обработки ФЭС XML файлов в Excel-отчет и протокол с доступом к файлам через SAMBA (`rail_connectors`).
[![linting: ruff](https://img.shields.io/badge/linting-ruff-f9c700)](https://github.com/astral-sh/ruff)
__________________________________________________________________________________________________________
# sfm, v 0.0.1
__sfm__
| | Исполнитель | Заказчик |
|-----------|:----------------------------------------:|:----------------------------------------------:|
| Сотрудник | raykov-mse | raykov-mse |
| Отдел | aurora | aurora |
| Контакты | raykovmse@rshb.ru | raykovmse@rshb.ru |
-----
## Архитектура
```text
main.py
app/
app_config.py # константы приложения и user-id
path_store.py # JSON-хранилище персональных путей
path_settings.py # UI-модалка изменения пути
samba_access.py # проверка доступа к SAMBA папке
config.py # переменные окружения с префиксом OPENBAO__SETTINGS__
ui.py # ui-компоненты
auth/
service.py # orchestration auth/session
providers.py
mock_provider.py
pipeline/
config.py # шаблоны имен/путей пайплайна
file_manager.py # scan/move/upload/download в SAMBA
parser.py # namespace-aware парсер ФЭС XML
report.py # генерация report.xlsx
protocol.py # генерация protocol.xlsx
service.py # run_batch orchestration
tests/
unit/
smoke/
providers.py # local/mock и prod/raisa провайдеры
mock_provider.py # мок для локального запуска
```
## Основной поток
## Режимы запуска
1. Пользователь вводит SAMBA логин/пароль.
2. При необходимости меняет персональный `input_dir` (с pre-check доступа).
3. Нажимает `Запустить обработку xml в excel`.
4. Пайплайн:
- сканирует входную папку;
- невалидные имена переносит в `Err`;
- валидные XML парсит и формирует строки отчета;
- создает `report.xlsx`, `protocol.xlsx`, `process.log`;
- сохраняет артефакты в `Out/YYYY/MM_YYYY/YYYYMMDD/run_number/`;
- успешные файлы переносит в `Processed`.
5. В UI отображаются счетчики и пути к артефактам.
- `OPENBAO__SETTINGS__LOCATION=local` -> локальный mock вход через форму.
- `OPENBAO__SETTINGS__LOCATION=prod` (или пусто) -> вход через `raisa_streamlit_oauth`.
## Переменные окружения
## Важно по raisa-streamlit-oauth
Используются переменные авторизации Raisa (`OPENBAO__SETTINGS__*`).
SMBA source и дефолтный путь входа задаются в коде (`app/app_config.py`).
- Используется версия `raisa-streamlit-oauth==1.0.3`.
- До старта приложения выполняется `load_dotenv()`.
- Переменные `OPENBAO__SETTINGS__*` автоматически маппятся в обычные
(`AUTHORIZE_URL`, `TOKEN_URL`, `APP_NAMESPACE` и т.д.), чтобы библиотека
работала по каноничному шаблону из документации.
## Быстрый старт
@ -56,24 +45,9 @@ copy .env.example .env
streamlit run main.py
```
## Тесты и качество
## Линт и формат
```bash
pytest
ruff check .
black --check .
isort --check-only .
ruff format .
```
## Pre-commit
```bash
pre-commit install
pre-commit run --all-files
```
Хуки:
- `black`
- `isort`
- `ruff`
- `ruff-format`

1
app/__init__.py
View File

@ -1 +1,2 @@
"""Пакет приложения SFM Streamlit."""

26
app/app_config.py
View File

@ -1,26 +0,0 @@
from __future__ import annotations
from pathlib import Path
import streamlit as st
SFM_SMB_BASE_PATH = "//sgo-fc01-r13/inbox-intech"
SFM_DEFAULT_INPUT_DIR = r"//sgo-fc01-r13/inbox-intech"
ACCESS_SUPPORT_CONTACT = "администратору СФМ"
USER_PATHS_KEY = "user_input_dirs"
EDIT_PATH_OPEN_KEY = "edit_path_open"
EDIT_PATH_VALUE_KEY = "edit_path_value"
PATH_STORE_FILE = Path(__file__).resolve().parents[1] / ".user_paths.json"
def resolve_user_id() -> str:
profile = st.session_state.get("user_profile", {})
if not isinstance(profile, dict):
return "anonymous|anonymous"
email = str(
profile.get("email") or profile.get("preferred_username") or "anonymous"
)
full_name = str(profile.get("name") or profile.get("full_name") or "anonymous")
return f"{email}|{full_name}"

1
app/auth/__init__.py
View File

@ -1 +1,2 @@
"""Вспомогательные модули аутентификации SFM Streamlit."""

1
app/auth/mock_provider.py
View File

@ -31,3 +31,4 @@ def get_jwt_token() -> dict[str, str] | None:
"realm_access": {"roles": ["local_dev"]},
}
return st.session_state["local_mock_profile"]

17
app/auth/providers.py
View File

@ -1,8 +1,9 @@
from __future__ import annotations
import os
from typing import Any, Protocol
from app.config import get_location
from .mock_provider import get_jwt_token as get_jwt_token_local
from .mock_provider import protect_application as protect_application_local
@ -19,11 +20,11 @@ class LocalMockProvider:
class RaisaProvider:
def get_token_data(self) -> dict[str, Any] | None:
from raisa_streamlit_oauth import (
get_jwt_token as get_jwt_token_raisa, # type: ignore[reportMissingImports]
from raisa_streamlit_oauth import ( # type: ignore[reportMissingImports]
get_jwt_token as get_jwt_token_raisa,
)
from raisa_streamlit_oauth import (
protect_application as protect_application_raisa, # type: ignore[reportMissingImports]
from raisa_streamlit_oauth import ( # type: ignore[reportMissingImports]
protect_application as protect_application_raisa,
)
protect_application_raisa(render_exit=False)
@ -31,9 +32,7 @@ class RaisaProvider:
def build_auth_provider() -> AuthProvider:
location = os.getenv(
"OPENBAO__SETTINGS__LOCATION", os.getenv("LOCATION", "prod")
).lower()
if location == "local":
if get_location() == "local":
return LocalMockProvider()
return RaisaProvider()

13
app/auth/service.py
View File

@ -12,13 +12,7 @@ def clear_session() -> None:
class AuthService:
def run(self) -> bool:
if "oauth_authenticated" not in st.session_state:
st.session_state["oauth_authenticated"] = False
if (
st.session_state.get("oauth_authenticated")
and "user_profile" in st.session_state
):
if "user_profile" in st.session_state:
return True
try:
@ -26,8 +20,7 @@ class AuthService:
except ImportError:
st.error(
"Библиотека raisa_streamlit_oauth недоступна. "
"Для локальной разработки установите LOCATION=local "
"или OPENBAO__SETTINGS__LOCATION=local."
"Для локальной разработки установите LOCATION=local."
)
return False
@ -44,5 +37,5 @@ class AuthService:
st.session_state["token_data"] = token_data
st.session_state["user_profile"] = token_data
st.session_state["oauth_authenticated"] = True
return True

43
app/config.py Normal file
View File

@ -0,0 +1,43 @@
from __future__ import annotations
import os
SETTINGS_PREFIX = "OPENBAO__SETTINGS__"
RAISA_ENV_KEYS = (
"AUTHORIZE_URL",
"TOKEN_URL",
"REFRESH_TOKEN_URL",
"LOGOUT_URL",
"JWKS_URL",
"CLIENT_ID",
"CLIENT_SECRET",
"REDIRECT_URI",
"SCOPE",
"APP_ID",
"APP_NAMESPACE",
"APP_NAME",
"LOCATION",
)
def get_setting(name: str) -> str:
prefixed = os.getenv(f"{SETTINGS_PREFIX}{name}", "").strip()
if prefixed:
return prefixed
return os.getenv(name, "").strip()
def get_location() -> str:
return get_setting("LOCATION").lower() or "prod"
def apply_raisa_env_aliases() -> None:
"""
Заполняет обычные имена переменных из OPENBAO__SETTINGS__,
чтобы raisa_streamlit_oauth работал одинаково в локальном и raisa runtime.
"""
for key in RAISA_ENV_KEYS:
prefixed = os.getenv(f"{SETTINGS_PREFIX}{key}", "").strip()
if prefixed and not os.getenv(key):
os.environ[key] = prefixed

65
app/path_settings.py
View File

@ -1,65 +0,0 @@
from __future__ import annotations
import streamlit as st
from app.app_config import (
EDIT_PATH_OPEN_KEY,
EDIT_PATH_VALUE_KEY,
SFM_DEFAULT_INPUT_DIR,
SFM_SMB_BASE_PATH,
)
from app.path_store import get_user_path
def ensure_user_path_state(user_id: str) -> str:
current_path = get_user_path(user_id)
st.session_state.setdefault(EDIT_PATH_OPEN_KEY, False)
return current_path
def open_path_editor(current_path: str) -> None:
st.session_state[EDIT_PATH_OPEN_KEY] = True
st.session_state[EDIT_PATH_VALUE_KEY] = current_path
def close_path_editor() -> None:
st.session_state[EDIT_PATH_OPEN_KEY] = False
def render_path_editor() -> str | None:
if not st.session_state.get(EDIT_PATH_OPEN_KEY):
return None
with st.container(border=True):
title_col, close_col = st.columns([9, 1])
with title_col:
st.markdown(
"**Введите путь по которому обрабатывать файлы только для вас**"
)
with close_col:
if st.button("✕", key="close_edit_path"):
close_path_editor()
st.rerun()
edited_path = st.text_input(
"Путь",
key=EDIT_PATH_VALUE_KEY,
placeholder=r"//sgo-fc01-r13/inbox-intech/Exchange/SMBDEMO/test",
help=(
"Укажите абсолютный UNC-путь к папке обработки, например "
f"`{SFM_SMB_BASE_PATH}/Exchange/SMBDEMO/test`."
),
)
save_col, cancel_col = st.columns(2)
with save_col:
if st.button("Сохранить", use_container_width=True):
return edited_path
with cancel_col:
if st.button("Отменить", use_container_width=True):
close_path_editor()
st.rerun()
return None
def get_resolved_input_dir(user_id: str) -> str:
return get_user_path(user_id) or SFM_DEFAULT_INPUT_DIR

43
app/path_store.py
View File

@ -1,43 +0,0 @@
from __future__ import annotations
import json
from app.app_config import PATH_STORE_FILE, SFM_DEFAULT_INPUT_DIR
def get_user_path(user_id: str) -> str:
payload = _load_data()
value = payload.get(user_id)
if isinstance(value, str) and value.strip():
return value
return SFM_DEFAULT_INPUT_DIR
def set_user_path(user_id: str, input_dir: str) -> None:
payload = _load_data()
payload[user_id] = input_dir
_save_data(payload)
def _load_data() -> dict[str, str]:
if not PATH_STORE_FILE.exists():
return {}
try:
raw = json.loads(PATH_STORE_FILE.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return {}
if not isinstance(raw, dict):
return {}
normalized: dict[str, str] = {}
for key, value in raw.items():
if isinstance(key, str) and isinstance(value, str):
normalized[key] = value
return normalized
def _save_data(payload: dict[str, str]) -> None:
PATH_STORE_FILE.parent.mkdir(parents=True, exist_ok=True)
PATH_STORE_FILE.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True),
encoding="utf-8",
)

1
app/pipeline/__init__.py
View File

@ -1 +0,0 @@
"""Пайплайн обработки XML в Excel."""

87
app/pipeline/config.py
View File

@ -1,87 +0,0 @@
from __future__ import annotations
import re
from datetime import datetime
FILE_NAME_PATTERNS = (
re.compile(r"^SKO115FZ_\d{2}_[A-Za-zА-Яа-я0-9]{9}_\d{8}_(?:[ХX]\d{5}|\d{6})\.xml$"),
)
DEFAULT_OUTPUT_DIR_NAME = "Out"
DEFAULT_ERROR_DIR_NAME = "Err"
DEFAULT_PROCESSED_DIR_NAME = "Processed"
REPORT_FILE_TEMPLATE = "report_{date_key}_{run_number}.xlsx"
PROTOCOL_FILE_TEMPLATE = "protocol_{date_key}_{run_number}.xlsx"
PROCESS_LOG_FILE_TEMPLATE = "process_{date_key}_{run_number}.log"
def validate_file_name(file_name: str) -> bool:
return any(pattern.match(file_name) for pattern in FILE_NAME_PATTERNS)
def build_date_parts(now: datetime | None = None) -> tuple[str, str, str, str]:
current = now or datetime.now()
year = str(current.year)
month = str(current.month)
day = str(current.day)
date_key = current.strftime("%Y%m%d")
return year, month, day, date_key
def build_out_dir(base_dir: str, now: datetime | None = None) -> str:
year, month, day, _ = build_date_parts(now)
return join_path(base_dir, DEFAULT_OUTPUT_DIR_NAME, year, month, day)
def build_run_dir(base_dir: str, run_number: int, now: datetime | None = None) -> str:
out_dir = build_out_dir(base_dir, now)
return join_path(out_dir, str(run_number))
def build_error_dir(base_dir: str) -> str:
return join_path(base_dir, DEFAULT_ERROR_DIR_NAME)
def build_processed_dir(base_dir: str) -> str:
return join_path(base_dir, DEFAULT_PROCESSED_DIR_NAME)
def build_report_file_name(date_key: str, run_number: int) -> str:
return REPORT_FILE_TEMPLATE.format(date_key=date_key, run_number=run_number)
def build_protocol_file_name(date_key: str, run_number: int) -> str:
return PROTOCOL_FILE_TEMPLATE.format(date_key=date_key, run_number=run_number)
def build_process_log_file_name(date_key: str, run_number: int) -> str:
return PROCESS_LOG_FILE_TEMPLATE.format(date_key=date_key, run_number=run_number)
def join_path(*parts: str) -> str:
raw_parts = [part for part in parts if part and part.strip("/\\")]
if not raw_parts:
return ""
first = raw_parts[0]
is_unc = (
first.startswith("//")
or first.startswith("\\\\")
or first.startswith("/")
or first.startswith("\\")
)
cleaned = [part.replace("/", "\\").strip("\\") for part in raw_parts]
joined = "\\".join(cleaned)
if is_unc:
return "\\\\" + joined.lstrip("\\")
return joined
def get_parent_dir(path: str) -> str:
normalized = path.rstrip("/\\")
if "\\" in normalized:
return normalized.rsplit("\\", maxsplit=1)[0]
if "/" in normalized:
return normalized.rsplit("/", maxsplit=1)[0]
return normalized

401
app/pipeline/file_manager.py
View File

@ -1,401 +0,0 @@
from __future__ import annotations
import tempfile
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse
from .config import (
build_date_parts,
build_error_dir,
build_out_dir,
build_processed_dir,
get_parent_dir,
join_path,
validate_file_name,
)
@dataclass
class RunContext:
run_dir: str
out_day_dir: str
error_dir: str
processed_dir: str
run_number: int
date_key: str
@dataclass
class RemoteFile:
path: str
name: str
@dataclass
class InputFileScanResult:
valid_files: list[RemoteFile]
invalid_files: list[RemoteFile]
class SMBClientConnection:
def __init__(
self, smb_base_path: str, user_params: dict[str, str] | None = None
) -> None:
self.smb_base_path = _normalize_unc_path(smb_base_path)
self.user_name = str((user_params or {}).get("user_name") or "")
self.password = str((user_params or {}).get("password") or "")
self._registered = False
def __enter__(self) -> SMBClientConnection:
self._ensure_registered()
return self
def __exit__(self, exc_type, exc, tb) -> bool: # noqa: ANN001
return False
def path_exists(self, path: str) -> bool:
smbclient = _import_smbclient()
full_path = self._resolve_path(path)
try:
smbclient.stat(full_path)
return True
except Exception: # noqa: BLE001
return False
def listdir(self, path: str) -> list[str]:
smbclient = _import_smbclient()
full_path = self._resolve_path(path)
# In some smbclient/smbprotocol versions scandir() returns a generator,
# not a context manager, so we iterate directly.
entries = smbclient.scandir(full_path)
return [entry.name for entry in entries]
def is_file(self, path: str) -> bool:
smbclient = _import_smbclient()
full_path = self._resolve_path(path)
try:
with smbclient.open_file(full_path, mode="rb"):
return True
except Exception: # noqa: BLE001
return False
def mkdir(self, path: str, parent: bool = False) -> None:
smbclient = _import_smbclient()
full_path = self._resolve_path(path)
if parent:
smbclient.makedirs(full_path, exist_ok=True)
return
smbclient.mkdir(full_path)
def download(
self,
remote_file_path: str | None = None,
local_file_path: str | None = None,
path_from: str | None = None,
path_to: str | None = None,
replace: bool = False,
) -> None:
remote = remote_file_path or path_from
local = local_file_path or path_to
if not remote or not local:
raise RuntimeError("Некорректные параметры download().")
target = Path(local)
if target.exists() and not replace:
raise FileExistsError(str(target))
target.parent.mkdir(parents=True, exist_ok=True)
with self.open(remote, mode="rb") as remote_file:
target.write_bytes(remote_file.read())
def upload(
self,
remote_file_path: str | None = None,
local_file_path: str | None = None,
path_from: str | None = None,
path_to: str | None = None,
replace: bool = False,
) -> None:
remote = remote_file_path or path_to
local = local_file_path or path_from
if not remote or not local:
raise RuntimeError("Некорректные параметры upload().")
local_path = Path(local)
if not local_path.exists():
raise FileNotFoundError(str(local_path))
full_remote = self._resolve_path(remote)
if self.path_exists(remote) and not replace:
raise FileExistsError(full_remote)
self.mkdir(get_parent_dir(remote), parent=True)
smbclient = _import_smbclient()
with smbclient.open_file(full_remote, mode="wb") as remote_file:
remote_file.write(local_path.read_bytes())
def copy(self, path_from: str, path_to: str, replace: bool = False) -> None:
if self.path_exists(path_to) and not replace:
raise FileExistsError(path_to)
with self.open(path_from, mode="rb") as source_file:
data = source_file.read()
self.mkdir(get_parent_dir(path_to), parent=True)
with self.open(path_to, mode="wb") as target_file:
target_file.write(data)
def delete(self, path: str) -> None:
smbclient = _import_smbclient()
full_path = self._resolve_path(path)
smbclient.remove(full_path)
def open(self, path: str, mode: str = "rb"):
smbclient = _import_smbclient()
full_path = self._resolve_path(path)
return smbclient.open_file(full_path, mode=mode)
def _ensure_registered(self) -> None:
if self._registered:
return
if not self.user_name:
raise RuntimeError("Введите логин SAMBA.")
smbclient = _import_smbclient()
server = _extract_server(self.smb_base_path)
try:
smbclient.register_session(
server=server,
username=self.user_name,
password=self.password,
)
except Exception as exc: # noqa: BLE001
raise RuntimeError(f"Ошибка подключения к SMB: {exc}") from exc
self._registered = True
def _resolve_path(self, path: str) -> str:
if path.startswith("//") or path.startswith("\\\\"):
return _normalize_unc_path(path)
if _looks_like_unc_without_prefix(path, self.smb_base_path):
return _normalize_unc_path("//" + path.lstrip("/\\"))
return _normalize_unc_path(join_path(self.smb_base_path, path))
def connect_samba(
smb_base_path: str, user_params: dict[str, str] | None = None
) -> SMBClientConnection:
return SMBClientConnection(smb_base_path=smb_base_path, user_params=user_params)
def check_input_dir_access(
input_dir: Path | str, samba_conn
) -> tuple[bool, str | None]:
return _check_samba_dir_access(str(input_dir), samba_conn, "Папка")
def check_network_dir_access(
network_dir: str | Path, samba_conn
) -> tuple[bool, str | None]:
path = Path(network_dir)
if not str(path).startswith("\\\\"):
return False, f"Путь не является сетевым UNC-путем: {path}"
return _check_samba_dir_access(str(path), samba_conn, "Сетевая папка")
def create_run_context(
input_dir: str, samba_conn, now: datetime | None = None
) -> RunContext:
# Артефакты формируем внутри указанного пользователем пути обработки.
base_dir = input_dir
out_day_dir = build_out_dir(base_dir, now)
error_dir = build_error_dir(base_dir)
processed_dir = build_processed_dir(base_dir)
ensure_dir_exists(out_day_dir, samba_conn)
ensure_dir_exists(error_dir, samba_conn)
ensure_dir_exists(processed_dir, samba_conn)
run_number = next_run_number(out_day_dir, samba_conn)
run_dir = join_path(out_day_dir, str(run_number))
ensure_dir_exists(run_dir, samba_conn)
_, _, _, date_key = build_date_parts(now)
return RunContext(
run_dir=run_dir,
out_day_dir=out_day_dir,
error_dir=error_dir,
processed_dir=processed_dir,
run_number=run_number,
date_key=date_key,
)
def next_run_number(out_day_dir: str, samba_conn) -> int:
with samba_conn:
if not samba_conn.path_exists(out_day_dir):
return 1
entries = samba_conn.listdir(out_day_dir)
numbers: list[int] = []
for entry in entries:
if str(entry).isdigit():
numbers.append(int(entry))
return max(numbers, default=0) + 1
def scan_input_files(input_dir: str, samba_conn) -> InputFileScanResult:
valid_files: list[RemoteFile] = []
invalid_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 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)
def move_file_to_error(
file_path: str, file_name: str, error_dir: str, samba_conn
) -> str:
target = _next_available_samba_target(error_dir, file_name, samba_conn)
_samba_move_file(file_path, target, samba_conn)
return target
def move_file_to_processed(
file_path: str, file_name: str, processed_dir: str, samba_conn
) -> str:
target = _next_available_samba_target(processed_dir, file_name, samba_conn)
_samba_move_file(file_path, target, samba_conn)
return target
def ensure_dir_exists(path: str, samba_conn) -> None:
with samba_conn:
if not samba_conn.path_exists(path):
samba_conn.mkdir(path, parent=True)
def read_remote_file_bytes(remote_path: str, samba_conn) -> bytes:
if hasattr(samba_conn, "download"):
with tempfile.TemporaryDirectory() as temp_dir:
local_path = Path(temp_dir) / Path(remote_path).name
with samba_conn:
try:
samba_conn.download(
remote_file_path=remote_path,
local_file_path=str(local_path),
replace=True,
)
except TypeError:
samba_conn.download(path_from=remote_path, path_to=str(local_path))
return local_path.read_bytes()
if hasattr(samba_conn, "open"):
with samba_conn:
with samba_conn.open(remote_path, "rb") as remote_file:
return remote_file.read()
raise RuntimeError("Коннектор SAMBA не поддерживает чтение файла (download/open).")
def upload_local_file(local_path: Path, remote_path: str, samba_conn) -> None:
remote_dir = get_parent_dir(remote_path)
ensure_dir_exists(remote_dir, samba_conn)
if hasattr(samba_conn, "upload"):
with samba_conn:
try:
samba_conn.upload(
remote_file_path=remote_path,
local_file_path=str(local_path),
replace=True,
)
except TypeError:
samba_conn.upload(path_from=str(local_path), path_to=remote_path)
return
if hasattr(samba_conn, "copy"):
with samba_conn:
samba_conn.copy(
path_from=str(local_path), path_to=remote_path, replace=True
)
return
raise RuntimeError("Коннектор SAMBA не поддерживает запись файла (upload/copy).")
def _check_samba_dir_access(
target_dir: str, samba_conn, source_name: str
) -> tuple[bool, str | None]:
try:
with samba_conn:
if not samba_conn.path_exists(target_dir):
return False, f"{source_name} не найдена: {target_dir}"
_ = samba_conn.listdir(target_dir)
return True, None
except Exception as exc: # noqa: BLE001
return False, f"Ошибка доступа к папке: {target_dir}. {exc}"
def _list_samba_files(input_dir: str, samba_conn) -> list[str]:
items: list[str] = []
with samba_conn:
for name in sorted(samba_conn.listdir(input_dir)):
full_path = join_path(input_dir, name)
if not samba_conn.is_file(full_path):
continue
items.append(name)
return items
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):
samba_conn.mkdir(dir_path, parent=True)
base_target = join_path(dir_path, file_name)
if not samba_conn.path_exists(base_target):
return base_target
base = Path(file_name)
index = 1
while True:
candidate_name = f"{base.stem}_{index}{base.suffix}"
candidate = join_path(dir_path, candidate_name)
if not samba_conn.path_exists(candidate):
return candidate
index += 1
def _samba_move_file(path_from: str, path_to: str, samba_conn) -> None:
with samba_conn:
samba_conn.copy(path_from=path_from, path_to=path_to, replace=False)
samba_conn.delete(path_from)
def _extract_server(smb_base_path: str) -> str:
server = urlparse(smb_base_path).netloc
if not server:
raise RuntimeError(f"Некорректный SMB путь: {smb_base_path}")
return server
def _normalize_unc_path(path: str) -> str:
normalized = path.replace("\\", "/").strip()
while normalized.startswith("///"):
normalized = normalized[1:]
if not normalized.startswith("//"):
normalized = "//" + normalized.lstrip("/")
return normalized
def _import_smbclient():
try:
import smbclient # type: ignore[import-not-found]
except ImportError as exc: # noqa: BLE001
raise RuntimeError("Библиотека smbprotocol/smbclient не установлена.") from exc
return smbclient
def _looks_like_unc_without_prefix(path: str, smb_base_path: str) -> bool:
normalized_path = path.replace("/", "\\").lstrip("\\")
if not normalized_path:
return False
server = _extract_server(smb_base_path).lower()
return normalized_path.lower().startswith(f"{server}\\")

157
app/pipeline/mapping.py
View File

@ -1,157 +0,0 @@
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class MappingRule:
column_name: str
candidate_keys: tuple[str, ...]
source: str = "any"
MAPPING_DATA_FILE = Path(__file__).resolve().with_name("mapping_table.json")
TARGET_REPORT_COLUMNS_COUNT = 203
_MANUAL_COLUMN_NAME_FIXES: dict[int, str] = {
1: "Имя XML файла",
178: "Место государственной регистрации ЕИО/Бенефициара (структура полностью)",
183: "Номер",
}
_INVALID_TAGS = {"", "-", "Источник", "путь", "подразумеваются", "тэга", "нашла"}
def _load_mapping_data() -> list[dict[str, str | int]]:
raw = json.loads(MAPPING_DATA_FILE.read_text(encoding="utf-8"))
if not isinstance(raw, list):
return []
result: list[dict[str, str | int]] = []
for item in raw:
if not isinstance(item, dict):
continue
idx = int(item.get("index", 0))
if not 1 <= idx <= TARGET_REPORT_COLUMNS_COUNT:
continue
column_name = str(item.get("column_name", "")).strip()
xml_tag = str(item.get("xml_tag", "")).strip()
xml_path = str(item.get("xml_path", "")).strip()
if idx in _MANUAL_COLUMN_NAME_FIXES:
column_name = _MANUAL_COLUMN_NAME_FIXES[idx]
column_name = _cleanup_column_name(column_name)
result.append(
{
"index": idx,
"column_name": column_name,
"xml_tag": xml_tag,
"xml_path": xml_path,
}
)
return sorted(result, key=lambda row: int(row["index"]))
def _cleanup_column_name(value: str) -> str:
cleaned = re.sub(r"\s+", " ", value).strip()
cleaned = re.sub(r"\s+не\s+нашла.*$", "", cleaned, flags=re.IGNORECASE)
return cleaned.strip(" -")
def _build_fixed_columns(mapping_data: list[dict[str, str | int]]) -> tuple[str, ...]:
columns: list[str] = []
for row in mapping_data:
column = str(row.get("column_name", "")).strip()
columns.append(column or f"Колонка {row['index']}")
if len(columns) < TARGET_REPORT_COLUMNS_COUNT:
for index in range(len(columns) + 1, TARGET_REPORT_COLUMNS_COUNT + 1):
columns.append(f"Колонка {index}")
return tuple(columns[:TARGET_REPORT_COLUMNS_COUNT])
def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule, ...]:
rules: list[MappingRule] = []
participant_columns = {
"Признак резидента участника",
"Тип участника",
"ИНН участника",
"Наименование участника",
"Счет участника",
}
for row in mapping_data:
column = str(row.get("column_name", "")).strip()
tag = str(row.get("xml_tag", "")).strip()
path = str(row.get("xml_path", "")).strip()
if not column or tag in _INVALID_TAGS:
continue
candidates = [tag]
if column.startswith("ИНН "):
candidates = ["ИННФЛ", "ИННЮЛ", "ИНН", *candidates]
if column == "Тип участника":
candidates = ["Тип", "ТипУчастника", *candidates]
if path:
path_tag = path.strip("/").split("/")[-1]
if path_tag and path_tag not in candidates:
candidates.append(path_tag)
rules.append(
MappingRule(
column_name=column,
candidate_keys=tuple(candidates),
source=(
"participant"
if column in participant_columns or column.startswith("ИНН ")
else "any"
),
)
)
return tuple(rules)
_MAPPING_DATA = _load_mapping_data()
FIXED_REPORT_COLUMNS: tuple[str, ...] = _build_fixed_columns(_MAPPING_DATA)
REPORT_MAPPING_RULES: tuple[MappingRule, ...] = _build_rules(_MAPPING_DATA)
def build_fixed_row(
*,
file_name: str,
record_id: str,
operation_index: int,
operation_fields: dict[str, str],
participant_fields: dict[str, str],
) -> dict[str, str]:
row: dict[str, str] = {
"Имя XML файла": file_name,
"Идентификатор записи": record_id,
"Уникальный номер операции": str(operation_index),
}
merged = _merge_fields(operation_fields, participant_fields)
for rule in REPORT_MAPPING_RULES:
if row.get(rule.column_name):
continue
source_payload = participant_fields if rule.source == "participant" else merged
row[rule.column_name] = _pick_value(source_payload, rule.candidate_keys)
for column in FIXED_REPORT_COLUMNS:
row.setdefault(column, "")
return row
def _merge_fields(
operation_fields: dict[str, str],
participant_fields: dict[str, str],
) -> dict[str, str]:
merged: dict[str, str] = {}
merged.update(operation_fields)
merged.update(participant_fields)
return merged
def _pick_value(payload: dict[str, str], candidates: tuple[str, ...]) -> str:
for key in candidates:
if key in payload and payload[key]:
return payload[key]
short = key.split(".")[-1]
if short in payload and payload[short]:
return payload[short]
return ""

1829
app/pipeline/mapping_table.json
View File

File diff suppressed because it is too large Load Diff

238
app/pipeline/parser.py
View File

@ -1,238 +0,0 @@
from __future__ import annotations
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable
from .config import validate_file_name as validate_pipeline_file_name
@dataclass
class ParticipantRow:
file_name: str
operation_index: int
record_id: str
operation_fields: dict[str, str]
participant_fields: dict[str, str]
@dataclass
class OperationParseError:
operation_index: int
record_id: str
reason: str
@dataclass
class FileParseResult:
file_name: str
rows: list[ParticipantRow] = field(default_factory=list)
total_operations: int = 0
operation_errors: list[OperationParseError] = field(default_factory=list)
fatal_error: str | None = None
@property
def is_fatal(self) -> bool:
return self.fatal_error is not None
def validate_file_name(file_name: str) -> bool:
return validate_pipeline_file_name(file_name)
def parse_xml_file(file_path: Path) -> FileParseResult:
try:
content = file_path.read_bytes()
except OSError as exc:
result = FileParseResult(file_name=file_path.name)
result.fatal_error = f"Ошибка чтения файла: {exc}"
return result
return parse_xml_content(file_name=file_path.name, xml_content=content)
def parse_xml_content(file_name: str, xml_content: bytes | str) -> FileParseResult:
result = FileParseResult(file_name=file_name)
if not validate_file_name(file_name):
result.fatal_error = "Некорректное имя XML-файла"
return result
try:
root = ET.fromstring(xml_content)
except ET.ParseError as exc:
result.fatal_error = f"Ошибка XML: {exc}"
return result
try:
operations = _get_operations(root)
common_fields = _extract_common_fields(root)
except ValueError as exc:
result.fatal_error = str(exc)
return result
for index, operation in enumerate(operations, start=1):
result.total_operations += 1
record_id = f"ОПЕРАЦИЯ_{index}"
try:
operation_fields = _extract_direct_fields(
operation, excluded_tags={"УчастникОп"}
)
operation_fields = {**common_fields, **operation_fields}
record_id = _extract_record_id(operation_fields, index)
participants = _find_children_by_name(operation, "УчастникОп")
if participants:
for participant in participants:
participant_fields = _extract_direct_fields(participant)
result.rows.append(
ParticipantRow(
file_name=file_name,
operation_index=index,
record_id=record_id,
operation_fields=operation_fields,
participant_fields=participant_fields,
)
)
else:
result.rows.append(
ParticipantRow(
file_name=file_name,
operation_index=index,
record_id=record_id,
operation_fields=operation_fields,
participant_fields={},
)
)
except Exception as exc: # noqa: BLE001
result.operation_errors.append(
OperationParseError(
operation_index=index,
record_id=record_id,
reason=f"Ошибка обработки операции: {exc}",
)
)
return result
def _get_operations(root: ET.Element) -> list[ET.Element]:
if _normalize_tag(root.tag) != "СообщОперКО":
raise ValueError("Некорректный корневой элемент XML (ожидался СообщОперКО)")
inform_part = _find_child_by_name(root, "ИнформЧасть")
if inform_part is None:
raise ValueError("В XML отсутствует элемент ИнформЧасть")
details = _find_child_by_name(inform_part, "СведКО")
if details is None:
raise ValueError("В XML отсутствует элемент СведКО")
return _find_children_by_name(details, "Операция")
def _extract_direct_fields(
element: ET.Element,
excluded_tags: Iterable[str] | None = None,
) -> dict[str, str]:
excluded = set(excluded_tags or ())
fields: dict[str, str] = {}
for child in element:
tag = _normalize_tag(child.tag)
if tag in excluded:
continue
_collect_leaf_fields(child, tag, fields)
return fields
def _extract_record_id(operation_fields: dict[str, str], operation_index: int) -> str:
possible_keys = (
"ИдентификаторЗаписи",
"НомерЗаписи",
"ИдЗаписи",
"ИдЗап",
"ИдОпер",
"ID",
)
for key in possible_keys:
if operation_fields.get(key):
return operation_fields[key]
return f"ОПЕРАЦИЯ_{operation_index}"
def _normalize_tag(tag: str) -> str:
if "}" in tag:
return tag.split("}", maxsplit=1)[1]
return tag
def _find_child_by_name(parent: ET.Element, child_name: str) -> ET.Element | None:
for child in parent:
if _normalize_tag(child.tag) == child_name:
return child
return None
def _find_children_by_name(parent: ET.Element, child_name: str) -> list[ET.Element]:
return [child for child in parent if _normalize_tag(child.tag) == child_name]
def _extract_common_fields(root: ET.Element) -> dict[str, str]:
common: dict[str, str] = {}
service_part = _find_child_by_name(root, "СлужЧасть")
if service_part is not None:
common.update(_extract_direct_fields(service_part))
inform_part = _find_child_by_name(root, "ИнформЧасть")
if inform_part is None:
return common
info_bank = _find_child_by_name(inform_part, "ИнфБанк")
if info_bank is not None:
common.update(_extract_direct_fields(info_bank))
details = _find_child_by_name(inform_part, "СведКО")
if details is None:
return common
common.update(
_extract_direct_fields(details, excluded_tags={"Операция", "ИнфФилиал"})
)
branch_info = _find_child_by_name(details, "ИнфФилиал")
if branch_info is not None:
common.update(_extract_direct_fields(branch_info))
return common
def _collect_leaf_fields(
element: ET.Element, current_path: str, fields: dict[str, str]
) -> None:
children = list(element)
if not children:
_append_value(fields, current_path, (element.text or "").strip())
return
for child in children:
child_name = _normalize_tag(child.tag)
_collect_leaf_fields(child, f"{current_path}.{child_name}", fields)
def _append_value(fields: dict[str, str], key: str, value: str) -> None:
if key not in fields:
fields[key] = value
return
existing = fields[key]
if not value:
return
if not existing:
fields[key] = value
return
if value in existing.split("; "):
return
fields[key] = f"{existing}; {value}"

91
app/pipeline/protocol.py
View File

@ -1,91 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
@dataclass
class ProtocolRow:
file_name: str
status: str
message: str
operations_total: int
rows_written: int
launcher: str
processed_at: datetime
def write_protocol(rows: Iterable[ProtocolRow], destination: Path) -> None:
row_list = list(rows)
workbook = Workbook()
details = workbook.active
details.title = "Протокол"
details.freeze_panes = "A2"
columns = [
"Имя файла",
"Статус",
"Комментарий",
"Операций",
"Строк отчета",
"Запустил",
"Время обработки",
]
_write_header(details, columns)
for row_index, row in enumerate(row_list, start=2):
values = [
row.file_name,
row.status,
row.message,
row.operations_total,
row.rows_written,
row.launcher,
row.processed_at.strftime("%Y-%m-%d %H:%M:%S"),
]
for col_index, value in enumerate(values, start=1):
details.cell(row=row_index, column=col_index, value=value)
_auto_width(details, columns)
_write_summary(workbook, row_list)
destination.parent.mkdir(parents=True, exist_ok=True)
workbook.save(destination)
def _write_summary(workbook: Workbook, rows: list[ProtocolRow]) -> None:
summary = workbook.create_sheet("Итог")
total = len(rows)
success = sum(1 for row in rows if row.status == "success")
partial = sum(1 for row in rows if row.status == "partial")
errors = sum(1 for row in rows if row.status == "error")
metrics = [
("Всего файлов", total),
("Успешно", success),
("Частично", partial),
("Ошибки", errors),
]
for index, (name, value) in enumerate(metrics, start=1):
summary.cell(row=index, column=1, value=name).font = Font(bold=True)
summary.cell(row=index, column=2, value=value)
summary.column_dimensions[get_column_letter(1)].width = 24
summary.column_dimensions[get_column_letter(2)].width = 14
def _write_header(sheet, columns: list[str]) -> None:
fill = PatternFill(start_color="D9E1F2", end_color="D9E1F2", fill_type="solid")
for index, name in enumerate(columns, start=1):
cell = sheet.cell(row=1, column=index, value=name)
cell.font = Font(bold=True)
cell.alignment = Alignment(horizontal="center", vertical="center")
cell.fill = fill
def _auto_width(sheet, columns: list[str]) -> None:
for index, name in enumerate(columns, start=1):
width = max(14, min(70, len(name) + 6))
sheet.column_dimensions[get_column_letter(index)].width = width

69
app/pipeline/report.py
View File

@ -1,69 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from .mapping import FIXED_REPORT_COLUMNS, build_fixed_row
@dataclass
class ReportRow:
file_name: str
record_id: str
operation_index: int
operation_fields: dict[str, str]
participant_fields: dict[str, str]
def write_report(rows: Iterable[ReportRow], destination: Path) -> None:
row_list = list(rows)
columns = list(FIXED_REPORT_COLUMNS)
workbook = Workbook()
sheet = workbook.active
sheet.title = "Отчет"
sheet.freeze_panes = "A2"
_write_header(sheet, columns)
for row_index, row in enumerate(row_list, start=2):
data = build_fixed_row(
file_name=row.file_name,
record_id=row.record_id,
operation_index=row.operation_index,
operation_fields=row.operation_fields,
participant_fields=row.participant_fields,
)
for col_index, column_name in enumerate(columns, start=1):
cell = sheet.cell(
row=row_index, column=col_index, value=data.get(column_name, "")
)
if "Дата" in column_name:
cell.number_format = "DD.MM.YYYY"
if "Сумм" in column_name or "Сумма" in column_name:
cell.number_format = "#,##0.00"
_auto_width(sheet, columns)
destination.parent.mkdir(parents=True, exist_ok=True)
workbook.save(destination)
def _write_header(sheet, columns: list[str]) -> None:
fill = PatternFill(start_color="E2F0D9", end_color="E2F0D9", fill_type="solid")
for index, name in enumerate(columns, start=1):
cell = sheet.cell(row=1, column=index, value=name)
cell.font = Font(bold=True)
cell.alignment = Alignment(
horizontal="center", vertical="center", wrap_text=True
)
cell.fill = fill
def _auto_width(sheet, columns: list[str]) -> None:
for index, column_name in enumerate(columns, start=1):
width = max(12, min(60, len(column_name) + 4))
column_letter = get_column_letter(index)
sheet.column_dimensions[column_letter].width = width

289
app/pipeline/service.py
View File

@ -1,289 +0,0 @@
from __future__ import annotations
import tempfile
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from .config import (
build_process_log_file_name,
build_protocol_file_name,
build_report_file_name,
join_path,
)
from .file_manager import (
check_input_dir_access,
connect_samba,
create_run_context,
move_file_to_error,
move_file_to_processed,
read_remote_file_bytes,
scan_input_files,
upload_local_file,
)
from .parser import FileParseResult, parse_xml_content
from .protocol import ProtocolRow, write_protocol
from .report import ReportRow, write_report
EMPTY_BATCH_STATUS = "info"
@dataclass
class BatchResult:
status: str
processed: int
partial: int
errors: int
report_path: str
protocol_path: str
log_path: str
messages: list[str] = field(default_factory=list)
def run_batch(
smb_base_path: str,
input_dir: str,
samba_user: str,
samba_password: str,
launcher: str,
) -> BatchResult:
messages: list[str] = []
if not samba_user:
return BatchResult(
status="error",
processed=0,
partial=0,
errors=1,
report_path="",
protocol_path="",
log_path="",
messages=["Введите логин SAMBA."],
)
samba_conn = connect_samba(
smb_base_path=smb_base_path,
user_params={"user_name": samba_user, "password": samba_password},
)
is_ok, access_error = check_input_dir_access(
input_dir=input_dir, samba_conn=samba_conn
)
if not is_ok:
return BatchResult(
status="error",
processed=0,
partial=0,
errors=1,
report_path="",
protocol_path="",
log_path="",
messages=[access_error or "Ошибка доступа к входной папке."],
)
run_context = create_run_context(input_dir=input_dir, samba_conn=samba_conn)
report_rows: list[ReportRow] = []
protocol_rows: list[ProtocolRow] = []
log_lines: list[str] = []
scan_result = scan_input_files(input_dir=input_dir, samba_conn=samba_conn)
log_lines.append(
f"{_now()} | scan files: valid={len(scan_result.valid_files)} invalid={len(scan_result.invalid_files)} "
)
for remote_file in scan_result.invalid_files:
reason = "Некорректное имя файла"
moved_to = move_file_to_error(
file_path=remote_file.path,
file_name=remote_file.name,
error_dir=run_context.error_dir,
samba_conn=samba_conn,
)
protocol_rows.append(
ProtocolRow(
file_name=remote_file.name,
status="error",
message=reason,
operations_total=0,
rows_written=0,
launcher=launcher,
processed_at=datetime.now(),
)
)
if moved_to:
log_lines.append(
f"{_now()} | invalid -> err: {remote_file.path} => {moved_to}"
)
for remote_file in scan_result.valid_files:
parse_result = _parse_remote_xml(remote_file.name, remote_file.path, samba_conn)
report_rows.extend(_report_rows_from_parse(parse_result))
status, message = _resolve_parse_status(parse_result)
if status == "success":
try:
moved_to = move_file_to_processed(
file_path=remote_file.path,
file_name=remote_file.name,
processed_dir=run_context.processed_dir,
samba_conn=samba_conn,
)
log_lines.append(
f"{_now()} | success -> processed: {remote_file.path} => {moved_to}"
)
except Exception as exc: # noqa: BLE001
status = "partial"
message = f"{message} Не удалось переместить в Processed: {exc}"
log_lines.append(
f"{_now()} | move to Processed failed: {remote_file.path}. {exc}"
)
else:
try:
moved_to = move_file_to_error(
file_path=remote_file.path,
file_name=remote_file.name,
error_dir=run_context.error_dir,
samba_conn=samba_conn,
)
log_lines.append(
f"{_now()} | {status} -> err: {remote_file.path} => {moved_to}"
)
except Exception as exc: # noqa: BLE001
message = f"{message} Не удалось переместить в Err: {exc}"
if status == "success":
status = "partial"
log_lines.append(
f"{_now()} | move to Err failed: {remote_file.path}. {exc}"
)
protocol_rows.append(
ProtocolRow(
file_name=remote_file.name,
status=status,
message=message,
operations_total=parse_result.total_operations,
rows_written=len(parse_result.rows),
launcher=launcher,
processed_at=datetime.now(),
)
)
if not protocol_rows:
protocol_rows.append(
ProtocolRow(
file_name="-",
status=EMPTY_BATCH_STATUS,
message="Новых файлов для обработки нет.",
operations_total=0,
rows_written=0,
launcher=launcher,
processed_at=datetime.now(),
)
)
log_lines.append(f"{_now()} | empty batch")
report_file_name = build_report_file_name(
run_context.date_key, run_context.run_number
)
protocol_file_name = build_protocol_file_name(
run_context.date_key, run_context.run_number
)
log_file_name = build_process_log_file_name(
run_context.date_key, run_context.run_number
)
report_path = join_path(run_context.run_dir, report_file_name)
protocol_path = join_path(run_context.run_dir, protocol_file_name)
log_path = join_path(run_context.run_dir, log_file_name)
_upload_artifacts(
report_rows=report_rows,
protocol_rows=protocol_rows,
log_lines=log_lines,
report_path=report_path,
protocol_path=protocol_path,
log_path=log_path,
samba_conn=samba_conn,
)
processed = sum(1 for row in protocol_rows if row.status == "success")
partial = sum(1 for row in protocol_rows if row.status == "partial")
errors = sum(1 for row in protocol_rows if row.status == "error")
status = (
"success"
if errors == 0 and partial == 0
else ("partial" if partial > 0 else "error")
)
messages.extend(log_lines[-5:])
return BatchResult(
status=status,
processed=processed,
partial=partial,
errors=errors,
report_path=report_path,
protocol_path=protocol_path,
log_path=log_path,
messages=messages,
)
def _parse_remote_xml(file_name: str, remote_path: str, samba_conn) -> FileParseResult:
try:
content = read_remote_file_bytes(remote_path=remote_path, samba_conn=samba_conn)
except Exception as exc: # noqa: BLE001
result = FileParseResult(file_name=file_name)
result.fatal_error = f"Ошибка чтения файла из SAMBA: {exc}"
return result
return parse_xml_content(file_name=file_name, xml_content=content)
def _report_rows_from_parse(parse_result: FileParseResult) -> list[ReportRow]:
result: list[ReportRow] = []
for row in parse_result.rows:
result.append(
ReportRow(
file_name=row.file_name,
record_id=row.record_id,
operation_index=row.operation_index,
operation_fields=row.operation_fields,
participant_fields=row.participant_fields,
)
)
return result
def _resolve_parse_status(parse_result: FileParseResult) -> tuple[str, str]:
if parse_result.fatal_error:
return "error", parse_result.fatal_error
if parse_result.operation_errors:
details = "; ".join(
f"операция {item.operation_index}: {item.reason}"
for item in parse_result.operation_errors
)
return "partial", details
return "success", "Успешно обработан."
def _upload_artifacts(
report_rows: list[ReportRow],
protocol_rows: list[ProtocolRow],
log_lines: list[str],
report_path: str,
protocol_path: str,
log_path: str,
samba_conn,
) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
temp_dir_path = Path(temp_dir)
local_report = temp_dir_path / "report.xlsx"
local_protocol = temp_dir_path / "protocol.xlsx"
local_log = temp_dir_path / "process.log"
write_report(report_rows, local_report)
write_protocol(protocol_rows, local_protocol)
# Use CRLF to keep one-record-per-line in Windows viewers.
local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8")
upload_local_file(local_report, report_path, samba_conn=samba_conn)
upload_local_file(local_protocol, protocol_path, samba_conn=samba_conn)
upload_local_file(local_log, log_path, samba_conn=samba_conn)
def _now() -> str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

42
app/samba_access.py
View File

@ -1,42 +0,0 @@
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

12
app/ui.py
View File

@ -2,6 +2,17 @@ from __future__ import annotations
import streamlit as st
from app.config import SETTINGS_PREFIX, get_location
def render_env_check() -> None:
with st.expander("Проверка env окружения"):
st.write(f"Текущий режим: `{get_location()}`")
st.code(
f"{SETTINGS_PREFIX}LOCATION=local # для локального mock-режима",
language="text",
)
def render_profile(profile: dict) -> None:
st.success("Аутентификация через Keycloak Raisa выполнена.")
@ -14,3 +25,4 @@ def render_profile(profile: dict) -> None:
"realm_roles": profile.get("realm_access", {}).get("roles", []),
}
)

123
main.py
View File

@ -1,137 +1,28 @@
import streamlit as st
from dotenv import load_dotenv
from app.app_config import ACCESS_SUPPORT_CONTACT, SFM_SMB_BASE_PATH, resolve_user_id
from app.auth.service import AuthService, clear_session
from app.path_settings import (
close_path_editor,
ensure_user_path_state,
get_resolved_input_dir,
open_path_editor,
render_path_editor,
)
from app.path_store import set_user_path
from app.pipeline.service import run_batch
from app.samba_access import check_access_and_list_files
from app.config import apply_raisa_env_aliases
from app.ui import render_env_check, render_profile
def main() -> None:
load_dotenv()
apply_raisa_env_aliases()
st.set_page_config(page_title="СФМ", page_icon=":lock:", layout="wide")
st.title("СФМ")
st.set_page_config(page_title="SFM Streamlit", page_icon=":lock:", layout="wide")
st.title("SFM Streamlit: Keycloak-only auth")
render_env_check()
if not AuthService().run():
return
_, center_col, _ = st.columns([1, 2, 1])
with center_col:
user_id = resolve_user_id()
current_input_dir = ensure_user_path_state(user_id)
resolved_input_dir = get_resolved_input_dir(user_id)
st.subheader("Подключение к сетевому диску")
samba_user = st.text_input(
"Логин SAMBA",
placeholder="Ivanov-IIv",
help="Логин учетной записи для доступа к SMB-шаре.",
)
samba_password = st.text_input(
"Пароль SAMBA",
type="password",
placeholder="12345abc",
help="Пароль учетной записи для доступа к SMB-шаре.",
)
if st.button("Изменить путь", use_container_width=True):
open_path_editor(current_input_dir)
edited_path = render_path_editor()
if edited_path is not None:
ok, message, _ = check_access_and_list_files(
smb_base_path=SFM_SMB_BASE_PATH,
input_dir=edited_path,
samba_user=samba_user,
samba_password=samba_password,
)
if ok:
set_user_path(user_id, edited_path)
close_path_editor()
st.success("Путь сохранен.")
st.rerun()
else:
st.error(_format_access_error(message))
resolved_input_dir = get_resolved_input_dir(user_id)
st.caption(f"Текущий путь: `{resolved_input_dir}`")
if st.button(
"Запустить обработку xml в excel",
type="primary",
use_container_width=True,
):
with st.spinner("Выполняется обработка..."):
launcher_name = samba_user.strip() or user_id
batch_result = run_batch(
smb_base_path=SFM_SMB_BASE_PATH,
input_dir=resolved_input_dir,
samba_user=samba_user,
samba_password=samba_password,
launcher=launcher_name,
)
if batch_result.status == "success":
st.success("Обработка завершена успешно.")
elif batch_result.status == "partial":
st.warning("Обработка завершена частично.")
else:
st.error("Обработка завершена с ошибками.")
st.write(
{
"processed": batch_result.processed,
"partial": batch_result.partial,
"errors": batch_result.errors,
"report_path": batch_result.report_path,
"protocol_path": batch_result.protocol_path,
"log_path": batch_result.log_path,
}
)
if batch_result.messages:
st.caption("Последние сообщения обработки:")
for message in batch_result.messages:
st.write(f"- {message}")
if st.button("Проверить доступ к папке", use_container_width=True):
ok, message, files = check_access_and_list_files(
smb_base_path=SFM_SMB_BASE_PATH,
input_dir=resolved_input_dir,
samba_user=samba_user,
samba_password=samba_password,
)
if not ok:
st.error(_format_access_error(message))
else:
if files:
st.success(f"Папка доступна. Найдено файлов: {len(files)}")
st.write(files)
else:
st.info("Папка доступна, но пуста.")
render_profile(st.session_state["user_profile"])
if st.button("Выйти", type="secondary"):
clear_session()
st.rerun()
def _format_access_error(message: str) -> str:
lowered = (message or "").lower()
user_input_keywords = (
"логин",
"парол",
"status_logon_failure",
"logon",
"authentication",
"auth",
)
if any(keyword in lowered for keyword in user_input_keywords):
return message
return f"{message} Обратитесь к {ACCESS_SUPPORT_CONTACT}."
if __name__ == "__main__":
main()

12
requirements.txt
View File

@ -1,10 +1,4 @@
streamlit==1.41.1
streamlit==1.46.1
raisa-streamlit-oauth==1.0.3
python-dotenv==1.1.1
ruff==0.12.7
black==24.4.2
isort==5.13.2
pre-commit==3.7.1
openpyxl
pytest
smbprotocol==1.15.0
python-dotenv==1.0.1
ruff==0.5.2

8
tests/conftest.py
View File

@ -1,8 +0,0 @@
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))

115
tests/smoke/test_batch_smoke.py
View File

@ -1,115 +0,0 @@
from __future__ import annotations
import shutil
from pathlib import Path
from app.pipeline.service import run_batch
class FakeSamba:
def __init__(self, root: Path) -> None:
self.root = root
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb): # noqa: ANN001
return False
def _to_local(self, remote_path: str) -> Path:
cleaned = remote_path.replace("/", "\\").strip("\\")
return self.root / Path(cleaned)
def path_exists(self, path: str) -> bool:
return self._to_local(path).exists()
def listdir(self, path: str) -> list[str]:
target = self._to_local(path)
if not target.exists():
return []
return [item.name for item in target.iterdir()]
def is_file(self, path: str) -> bool:
return self._to_local(path).is_file()
def mkdir(self, path: str, parent: bool = False) -> None:
target = self._to_local(path)
if parent:
target.mkdir(parents=True, exist_ok=True)
else:
target.mkdir(exist_ok=True)
def copy(self, path_from: str, path_to: str, replace: bool = False) -> None:
source = self._to_local(path_from)
target = self._to_local(path_to)
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists() and not replace:
raise FileExistsError(path_to)
shutil.copy2(source, target)
def delete(self, path: str) -> None:
target = self._to_local(path)
if target.exists():
target.unlink()
def open(self, path: str, mode: str):
return self._to_local(path).open(mode)
def upload(self, path_from: str, path_to: str) -> None:
source = Path(path_from)
target = self._to_local(path_to)
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
def _valid_xml() -> str:
return """<?xml version="1.0" encoding="UTF-8"?>
<СообщОперКО xmlns="urn:test">
<СлужЧасть><ИдФайл>ID1</ИдФайл></СлужЧасть>
<ИнформЧасть>
<СведКО>
<Операция>
<ИдентификаторЗаписи>OP1</ИдентификаторЗаписи>
<УчастникОп><Тип>01</Тип></УчастникОп>
</Операция>
</СведКО>
</ИнформЧасть>
</СообщОперКО>
"""
def test_run_batch_smoke_and_idempotency(tmp_path: Path, monkeypatch) -> None:
samba = FakeSamba(tmp_path)
in_dir_local = tmp_path / "Exchange" / "SMBDEMO" / "test"
in_dir_local.mkdir(parents=True)
(in_dir_local / "SKO115FZ_01_123456789_20260616_X00001.xml").write_text(
_valid_xml(),
encoding="utf-8",
)
from app.pipeline import service
monkeypatch.setattr(
service, "connect_samba", lambda smb_base_path, user_params: samba
)
first = run_batch(
smb_base_path="//test-server/share",
input_dir="Exchange\\SMBDEMO\\test",
samba_user="user",
samba_password="pass",
launcher="tester",
)
assert first.status in {"success", "partial"}
assert first.protocol_path
assert first.report_path
second = run_batch(
smb_base_path="//test-server/share",
input_dir="Exchange\\SMBDEMO\\test",
samba_user="user",
samba_password="pass",
launcher="tester",
)
assert second.status == "success"
assert second.processed == 0

114
tests/unit/test_file_manager.py
View File

@ -1,114 +0,0 @@
from __future__ import annotations
import shutil
from pathlib import Path
from app.pipeline.file_manager import (
create_run_context,
move_file_to_error,
next_run_number,
scan_input_files,
)
class FakeSamba:
def __init__(self, root: Path) -> None:
self.root = root
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb): # noqa: ANN001
return False
def _to_local(self, remote_path: str) -> Path:
cleaned = remote_path.replace("/", "\\").strip("\\")
return self.root / Path(cleaned)
def path_exists(self, path: str) -> bool:
return self._to_local(path).exists()
def listdir(self, path: str) -> list[str]:
target = self._to_local(path)
if not target.exists():
return []
return [item.name for item in target.iterdir()]
def is_file(self, path: str) -> bool:
return self._to_local(path).is_file()
def mkdir(self, path: str, parent: bool = False) -> None:
target = self._to_local(path)
if parent:
target.mkdir(parents=True, exist_ok=True)
else:
target.mkdir(exist_ok=True)
def copy(self, path_from: str, path_to: str, replace: bool = False) -> None:
source = self._to_local(path_from)
target = self._to_local(path_to)
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists() and not replace:
raise FileExistsError(path_to)
shutil.copy2(source, target)
def delete(self, path: str) -> None:
target = self._to_local(path)
if target.exists():
target.unlink()
def test_scan_input_files_splits_valid_and_invalid(tmp_path: Path) -> None:
samba = FakeSamba(tmp_path)
in_dir = "Exchange\\SMBDEMO\\test"
local_dir = tmp_path / "Exchange" / "SMBDEMO" / "test"
local_dir.mkdir(parents=True)
(local_dir / "SKO115FZ_01_123456789_20260616_X00001.xml").write_text(
"ok", encoding="utf-8"
)
(local_dir / "wrong_name.xml").write_text("bad", encoding="utf-8")
result = scan_input_files(in_dir, samba)
assert len(result.valid_files) == 1
assert len(result.invalid_files) == 1
def test_next_run_number_reads_existing_dirs(tmp_path: Path) -> None:
samba = FakeSamba(tmp_path)
out_day = (
tmp_path / "Exchange" / "SMBDEMO" / "Out" / "2026" / "06_2026" / "20260616"
)
(out_day / "1").mkdir(parents=True)
(out_day / "2").mkdir(parents=True)
number = next_run_number("Exchange\\SMBDEMO\\Out\\2026\\06_2026\\20260616", samba)
assert number == 3
def test_create_run_context_creates_paths(tmp_path: Path) -> None:
samba = FakeSamba(tmp_path)
context = create_run_context("Exchange\\SMBDEMO\\test", samba)
assert context.run_number == 1
assert (tmp_path / Path(context.run_dir)).exists()
assert (tmp_path / Path(context.error_dir)).exists()
assert (tmp_path / Path(context.processed_dir)).exists()
def test_move_file_to_error_adds_suffix_on_collision(tmp_path: Path) -> None:
samba = FakeSamba(tmp_path)
error_dir = tmp_path / "Exchange" / "SMBDEMO" / "Err"
error_dir.mkdir(parents=True)
(error_dir / "file.xml").write_text("existing", encoding="utf-8")
source = tmp_path / "Exchange" / "SMBDEMO" / "test" / "file.xml"
source.parent.mkdir(parents=True)
source.write_text("new", encoding="utf-8")
target = move_file_to_error(
file_path="Exchange\\SMBDEMO\\test\\file.xml",
file_name="file.xml",
error_dir="Exchange\\SMBDEMO\\Err",
samba_conn=samba,
)
assert target.endswith("file_1.xml")
assert (error_dir / "file_1.xml").exists()

71
tests/unit/test_mapping.py
View File

@ -1,71 +0,0 @@
from __future__ import annotations
from app.pipeline.mapping import FIXED_REPORT_COLUMNS, build_fixed_row
def _column_with(prefix: str) -> str:
for column in FIXED_REPORT_COLUMNS:
if column.startswith(prefix):
return column
raise AssertionError(f"Колонка с префиксом '{prefix}' не найдена")
def test_build_fixed_row_maps_known_fields() -> None:
row = build_fixed_row(
file_name="f.xml",
record_id="R1",
operation_index=2,
operation_fields={
"ИдФайл": "FILE_01",
"КодОперации": "5007",
"СумОперации": "100.55",
},
participant_fields={
"ИННФЛ": "123456789012",
"Тип": "01",
},
)
assert row["Имя XML файла"] == "f.xml"
assert row["Идентификатор записи"] == "R1"
assert row["Уникальный номер операции"] == "2"
assert row["Код вида операции"] == "5007"
assert row["Сумма в валюте проведения"] == "100.55"
assert row[_column_with("ИНН ")] == "123456789012"
assert row["Тип участника"] == "01"
def test_build_fixed_row_keeps_missing_values_empty() -> None:
row = build_fixed_row(
file_name="f.xml",
record_id="R1",
operation_index=1,
operation_fields={},
participant_fields={},
)
for column in FIXED_REPORT_COLUMNS:
assert column in row
assert row["Код вида операции"] == ""
assert row[_column_with("ИНН ")] == ""
def test_build_fixed_row_does_not_take_operation_inn_for_participant() -> None:
row = build_fixed_row(
file_name="f.xml",
record_id="R1",
operation_index=1,
operation_fields={
"ИНН": "0000000000",
"Тип": "99",
},
participant_fields={
"ИННФЛ": "111111111111",
"Тип": "01",
},
)
assert row[_column_with("ИНН ")] == "111111111111"
assert row["Тип участника"] == "01"
def test_fixed_report_columns_over_200() -> None:
assert len(FIXED_REPORT_COLUMNS) == 203

93
tests/unit/test_parser.py
View File

@ -1,93 +0,0 @@
from __future__ import annotations
from app.pipeline.parser import parse_xml_content
def _base_xml(operations: str) -> str:
return f"""<?xml version="1.0" encoding="UTF-8"?>
<СообщОперКО xmlns="urn:test">
<СлужЧасть><ИдФайл>ID1</ИдФайл></СлужЧасть>
<ИнформЧасть>
<ИнфБанк><БИК>123</БИК></ИнфБанк>
<СведКО>
{operations}
</СведКО>
</ИнформЧасть>
</СообщОперКО>
"""
def test_parse_xml_with_invalid_name_returns_fatal() -> None:
result = parse_xml_content("broken.xml", "<СообщОперКО/>")
assert result.is_fatal
assert "Некорректное имя XML-файла" in (result.fatal_error or "")
def test_parse_xml_with_malformed_xml_returns_fatal() -> None:
result = parse_xml_content("SKO115FZ_01_123456789_20260616_X00001.xml", "<broken")
assert result.is_fatal
assert "Ошибка XML" in (result.fatal_error or "")
def test_parse_xml_without_inform_part_returns_fatal() -> None:
xml = """<?xml version="1.0" encoding="UTF-8"?><СообщОперКО></СообщОперКО>"""
result = parse_xml_content("SKO115FZ_01_123456789_20260616_X00001.xml", xml)
assert result.is_fatal
assert "ИнформЧасть" in (result.fatal_error or "")
def test_parse_xml_without_participants_creates_single_row() -> None:
xml = _base_xml(
"<Операция><ИдентификаторЗаписи>OP1</ИдентификаторЗаписи><СуммаОпер>100</СуммаОпер></Операция>"
)
result = parse_xml_content("SKO115FZ_01_123456789_20260616_X00001.xml", xml)
assert not result.is_fatal
assert len(result.rows) == 1
assert result.rows[0].record_id == "OP1"
def test_parse_xml_with_namespace_and_two_participants() -> None:
xml = _base_xml(
"""
<Операция>
<ИдентификаторЗаписи>OP2</ИдентификаторЗаписи>
<УчастникОп><Тип>01</Тип></УчастникОп>
<УчастникОп><Тип>02</Тип></УчастникОп>
</Операция>
"""
)
result = parse_xml_content("SKO115FZ_01_123456789_20260616_X00001.xml", xml)
assert not result.is_fatal
assert len(result.rows) == 2
assert result.rows[0].participant_fields["Тип"] == "01"
assert result.rows[1].participant_fields["Тип"] == "02"
def test_parse_xml_operation_error_marks_partial(monkeypatch) -> None:
xml = _base_xml(
"<Операция><ИдентификаторЗаписи>OP3</ИдентификаторЗаписи><УчастникОп><Тип>01</Тип></УчастникОп></Операция>"
)
from app.pipeline import parser
original = parser._extract_direct_fields
def _boom(element, excluded_tags=None): # noqa: ANN001
tag_name = parser._normalize_tag(element.tag)
if tag_name == "УчастникОп":
raise ValueError("bad participant")
return original(element, excluded_tags=excluded_tags)
monkeypatch.setattr(parser, "_extract_direct_fields", _boom)
result = parse_xml_content("SKO115FZ_01_123456789_20260616_X00001.xml", xml)
assert not result.is_fatal
assert result.operation_errors
assert result.operation_errors[0].record_id == "OP3"
def test_parse_xml_with_numeric_suffix_file_name_is_valid() -> None:
xml = _base_xml(
"<Операция><ИдентификаторЗаписи>OP4</ИдентификаторЗаписи><СуммаОпер>100</СуммаОпер></Операция>"
)
result = parse_xml_content("SKO115FZ_01_044525111_20251224_000051.xml", xml)
assert not result.is_fatal
assert len(result.rows) == 1