105 lines
5.0 KiB
Python
105 lines
5.0 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
from tempfile import gettempdir
|
||
|
||
import streamlit as st
|
||
|
||
SFM_SMB_BASE_PATH = "//sgo-fc01-r06/sfm/ОТЧЕТ_7081У"
|
||
SFM_DEFAULT_INPUT_DIR = r"//sgo-fc01-r06/sfm/ОТЧЕТ_7081У"
|
||
ACCESS_SUPPORT_CONTACT = "администратору СФМ"
|
||
|
||
# TODO(BIQ-36593): подтвердить SPACE_ID для прода (сейчас — значение,
|
||
# проверенное вручную на тестовом пространстве "AURORA"). По наблюдениям
|
||
# ID может быть составным ("storageId$spaceId") — храним как есть, без
|
||
# разбора, через env-переменную, чтобы не хардкодить прод-значение в коде.
|
||
OPENCLOUD_SPACE_ID = os.getenv(
|
||
"SFM_OPENCLOUD_SPACE_ID", "28c0cecf-1e20-4664-b165-5c9d5515e70f"
|
||
)
|
||
|
||
EDIT_PATH_OPEN_KEY = "edit_path_open"
|
||
EDIT_PATH_VALUE_KEY = "edit_path_value"
|
||
# Runtime-only user paths storage. In bank contour we use tmpfs.
|
||
PATH_STORE_FILE = Path(gettempdir()) / "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}"
|
||
|
||
|
||
def resolve_opencloud_username() -> str | None:
|
||
"""preferred_username из Keycloak — им же заказчик регистрировал App Token
|
||
в веб-интерфейсе OpenCloud (сверено вручную, совпадает без домена/регистра).
|
||
|
||
Для локального теста (mock-провайдер) — fallback на .env.
|
||
"""
|
||
profile = st.session_state.get("user_profile", {})
|
||
if isinstance(profile, dict):
|
||
value = profile.get("preferred_username")
|
||
if value:
|
||
return str(value)
|
||
return os.getenv("SFM_OPENCLOUD_USERNAME") or None
|
||
|
||
|
||
def resolve_opencloud_access_token() -> str | None:
|
||
"""Достаёт App Token заказчика для авторизации в OpenCloud (Basic Auth).
|
||
|
||
ИСТОРИЯ (см. app/opencloud_access.py): изначально пытались использовать
|
||
Bearer ID-токен Keycloak-клиента datalab — Graph API OpenCloud отвечал
|
||
401 (несовпадение audience). Подтверждено на проде 16.09: рабочий путь —
|
||
HTTP Basic Auth с username (resolve_opencloud_username) и долгоживущим
|
||
App Token, который заказчик сам создаёт в веб-интерфейсе OpenCloud.
|
||
Токены хранятся зашифрованными в БД (app/opencloud_token_store.py),
|
||
заказчики присылают их нам сами при онбординге.
|
||
|
||
Приложение синхронное (Streamlit) — обёртка asyncio.run вокруг async
|
||
store оправдана только частотой вызова (раз на запуск обработки), при
|
||
росте нагрузки стоит пересмотреть.
|
||
"""
|
||
username = resolve_opencloud_username()
|
||
if not username:
|
||
return None
|
||
|
||
import asyncio
|
||
|
||
from app.opencloud_token_store import OpenCloudTokenStoreError, get_token
|
||
|
||
try:
|
||
return asyncio.run(get_token(username))
|
||
except OpenCloudTokenStoreError as exc:
|
||
st.error(f"Ошибка доступа к токену OpenCloud: {exc}")
|
||
return None
|
||
except Exception: # noqa: BLE001
|
||
# Локальный тест без поднятой БД/переменных окружения — fallback.
|
||
return os.getenv("SFM_OPENCLOUD_ACCESS_TOKEN") or None
|
||
|
||
|
||
def resolve_opencloud_user_folder() -> str | None:
|
||
"""Возвращает предполагаемое имя личной папки пользователя в OpenCloud.
|
||
|
||
ОТКРЫТЫЙ ВОПРОС: неизвестно, откуда в проде берётся имя папки вида
|
||
"Райков М.С." — это предположение, что claim "name" из Keycloak уже
|
||
приходит в этом формате (совпадает с тем, что руками использовалось в
|
||
open_cloud_upload_fixed.py). Альтернатива — папки создаются заранее
|
||
администратором по другому принципу (email/preferred_username), это
|
||
нужно сверить в проде.
|
||
|
||
Для локального теста — fallback на переменную окружения
|
||
SFM_OPENCLOUD_FOLDER из .env.
|
||
"""
|
||
profile = st.session_state.get("user_profile", {})
|
||
if isinstance(profile, dict):
|
||
value = profile.get("name")
|
||
if value:
|
||
return str(value)
|
||
return os.getenv("SFM_OPENCLOUD_FOLDER") or None
|