import asyncio import logging import shutil import tempfile from collections.abc import MutableMapping, Sequence from pathlib import Path import streamlit as st from dotenv import load_dotenv from app.app_config import ( resolve_opencloud_access_token, resolve_opencloud_username, resolve_user_id, ) from app.auth.service import AuthService, clear_session from app.opencloud_access import DriveQuota, resolve_personal_drive from app.opencloud_token_store import OpenCloudTokenStoreError, add_token from app.pipeline.service import BatchResult, run_batch logger = logging.getLogger(__name__) # Ключ загрузчика зависит от счётчика: Streamlit не умеет очищать # file_uploader программно, но виджет с новым ключом создаётся пустым. UPLOADER_GENERATION_KEY = "uploader_generation" # file_id вложений, которые пользователь исключил из обработки. Убрать файл # из самого file_uploader нельзя, поэтому он остаётся в загрузчике, но не # передаётся в run_batch. EXCLUDED_FILE_IDS_KEY = "excluded_file_ids" # Квота личного пространства (DriveQuota или None, если узнать не удалось). # Запрашивается один раз за сессию — см. _render_quota. QUOTA_STATE_KEY = "opencloud_quota" # Со скольких файлов список вложений получает свою прокрутку, чтобы партия # из ~100 файлов не растягивала страницу. _ATTACHMENT_LIST_SCROLL_FROM = 10 _BYTES_IN_GB = 1024**3 def main() -> None: load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s" ) st.set_page_config(page_title="СФМ", page_icon=":lock:", layout="wide") st.title("СФМ") if not AuthService().run(): return _init_attachment_state(st.session_state) _, center_col, _ = st.columns([1, 2, 1]) with center_col: _render_token_popover() _render_quota() st.caption( "Прикрепите XML-файлы и/или архивы с ними (можно несколько сразу, " "вперемешку) — отчёт, протокол и лог загрузятся в Out вашего " "личного пространства OpenCloud." ) uploaded_files = st.file_uploader( "XML-файлы и архивы", type=["xml", "zip", "tar", "gz", "tgz", "bz2", "xz"], accept_multiple_files=True, key=f"uploader_{st.session_state[UPLOADER_GENERATION_KEY]}", ) active_files = _render_attachment_list(uploaded_files or []) if st.button( "Запустить обработку xml в excel", type="primary", use_container_width=True, disabled=not active_files, ): with tempfile.TemporaryDirectory() as tmp_dir: input_paths = _save_attachments(active_files, Path(tmp_dir)) progress_bar = st.progress(0.0, text="Подготовка файлов...") status_placeholder = st.empty() def _on_progress(done: int, total: int, file_name: str) -> None: # Прогресс по файлам, а не один спиннер на весь батч — # иначе при большом количестве файлов (были жалобы на # батч в 230 штук) непонятно, идёт обработка или # приложение зависло, и пользователь закрывает вкладку # посреди обработки. fraction = done / total if total else 1.0 progress_bar.progress( fraction, text=f"Обработано {done} из {total}: {file_name}" ) status_placeholder.caption( f"Последний обработанный файл: {file_name}" ) launcher_name = resolve_opencloud_username() or resolve_user_id() batch_result = run_batch( input_paths=input_paths, launcher=launcher_name, opencloud_username=resolve_opencloud_username(), opencloud_access_token=resolve_opencloud_access_token(), progress_callback=_on_progress, ) progress_bar.progress(1.0, text="Готово") if batch_result.status == "success": st.success("Обработка завершена успешно.") elif batch_result.status == "partial": st.warning("Обработка завершена частично.") else: st.error("Обработка завершена с ошибками.") _render_result_links(batch_result) if batch_result.messages: st.caption("Последние сообщения обработки:") for message in batch_result.messages: st.write(f"- {message}") # Если обработка прошла, а загрузка в OpenCloud — нет, отдаём # результат напрямую, чтобы не терять долгую обработку. for file_name, payload in batch_result.fallback_files.items(): st.download_button( f"Скачать {file_name}", data=payload, file_name=file_name, use_container_width=True, key=f"fallback_{file_name}", ) if st.button("Выйти", type="secondary"): clear_session() st.rerun() def _init_attachment_state(state: MutableMapping) -> None: if UPLOADER_GENERATION_KEY not in state: state[UPLOADER_GENERATION_KEY] = 0 if EXCLUDED_FILE_IDS_KEY not in state: state[EXCLUDED_FILE_IDS_KEY] = set() def _clear_all_attachments(state: MutableMapping) -> None: """Сбрасывает загрузчик (новый ключ — пустой виджет) и список исключённых. Партии бывают по ~100 файлов, и удалять их по одному крестиком долго. Автоматически очищать загрузчик после обработки нельзя: к уже прикреплённым файлам иногда докидывают новые. """ state[UPLOADER_GENERATION_KEY] += 1 state[EXCLUDED_FILE_IDS_KEY] = set() def _exclude_checkbox_key(generation: int, file_id: str) -> str: return f"exclude_{generation}_{file_id}" def _active_uploads(uploaded_files: Sequence, excluded_file_ids: set[str]) -> list: """Вложения, которые пойдут в обработку. Файл определяется по file_id, а не по имени: имена вложений могут совпадать (два архива «Файлы.zip» из разных папок). """ return [ uploaded for uploaded in uploaded_files if uploaded.file_id not in excluded_file_ids ] def _exclude_selected(state: MutableMapping, active_files: Sequence) -> int: """Переносит отмеченные чекбоксами вложения в исключённые. Возвращает, сколько файлов исключено. """ generation = state[UPLOADER_GENERATION_KEY] selected = { uploaded.file_id for uploaded in active_files if state.get(_exclude_checkbox_key(generation, uploaded.file_id)) } if selected: state[EXCLUDED_FILE_IDS_KEY] = set(state[EXCLUDED_FILE_IDS_KEY]) | selected return len(selected) def _render_attachment_list(uploaded_files: Sequence) -> list: """Список прикреплённых файлов с групповым исключением и очисткой. Возвращает вложения, которые нужно передать в обработку. """ if not uploaded_files: return [] state = st.session_state generation = state[UPLOADER_GENERATION_KEY] active_files = _active_uploads(uploaded_files, state[EXCLUDED_FILE_IDS_KEY]) excluded_count = len(uploaded_files) - len(active_files) st.caption( f"В обработку пойдёт файлов: {len(active_files)} из {len(uploaded_files)}." ) if active_files: scroll_height = ( 300 if len(active_files) >= _ATTACHMENT_LIST_SCROLL_FROM else None ) with st.container(height=scroll_height, border=True): for uploaded in active_files: st.checkbox( f"{uploaded.name} ({_format_file_size(uploaded.size)})", key=_exclude_checkbox_key(generation, uploaded.file_id), ) exclude_col, clear_col = st.columns(2) with exclude_col: if st.button( "Исключить выбранные", use_container_width=True, disabled=not active_files, ): if _exclude_selected(state, active_files): st.rerun() st.info("Отметьте файлы, которые нужно исключить.") with clear_col: if st.button("Очистить все вложения", use_container_width=True): _clear_all_attachments(state) st.rerun() if excluded_count: st.caption( f"Исключено из обработки: {excluded_count}. Они остаются в " "загрузчике выше, но обрабатываться не будут." ) if st.button("Вернуть исключённые", key="restore_excluded"): state[EXCLUDED_FILE_IDS_KEY] = set() st.rerun() return active_files def _format_file_size(size_bytes: int) -> str: if size_bytes < 1024: return f"{size_bytes} Б" if size_bytes < 1024**2: return f"{size_bytes / 1024:.0f} КБ" return f"{size_bytes / 1024**2:.1f} МБ".replace(".", ",") def _render_quota() -> None: """«Занято X из Y ГБ» в личном пространстве OpenCloud. Квота запрашивается один раз за сессию (кеш в session_state, кнопка «Обновить» сбрасывает его). Если узнать её не удалось — ничего не показываем: это справочная информация, страница из-за неё падать не должна. """ state = st.session_state if QUOTA_STATE_KEY not in state: state[QUOTA_STATE_KEY] = _load_quota() text = _format_quota(state[QUOTA_STATE_KEY]) if not text: return text_col, refresh_col = st.columns([4, 1], vertical_alignment="center") text_col.caption(text) if refresh_col.button("Обновить", key="opencloud_quota_refresh"): state.pop(QUOTA_STATE_KEY, None) st.rerun() def _load_quota() -> DriveQuota | None: try: username = resolve_opencloud_username() token = resolve_opencloud_access_token() if username else None if not username or not token: return None return resolve_personal_drive(username, token).quota except Exception: # noqa: BLE001 logger.warning("Не удалось получить квоту OpenCloud", exc_info=True) return None def _format_quota(quota: DriveQuota | None) -> str | None: if quota is None: return None used = _format_gigabytes(quota.used) if quota.total: return f"Занято {used} из {_format_gigabytes(quota.total)} ГБ" return f"Занято {used} ГБ" def _format_gigabytes(size_bytes: int) -> str: value = size_bytes / _BYTES_IN_GB text = f"{value:.2f}" if value < 1 else f"{value:.1f}" return text.rstrip("0").rstrip(".").replace(".", ",") def _result_links(batch_result: BatchResult) -> list[tuple[str, str]]: """Кнопки-ссылки на результат: только те, для которых есть адрес.""" links = [ ("Открыть отчёт", batch_result.report_url), ("Протокол", batch_result.protocol_url), ("Лог", batch_result.log_url), ("Папка с результатами", batch_result.folder_url), ] return [(label, url) for label, url in links if url] def _render_result_links(batch_result: BatchResult) -> None: links = _result_links(batch_result) if not links: return for column, (label, url) in zip(st.columns(len(links)), links, strict=True): column.link_button(label, url, use_container_width=True) def _save_attachments(uploaded_files, target_dir: Path) -> list[Path]: """Складывает вложения на диск, каждое в свой подкаталог. Подкаталог по индексу нужен потому, что имена вложений могут совпадать (например, два архива "Файлы.zip" из разных папок), и одно перезаписало бы другое. """ paths: list[Path] = [] for index, uploaded in enumerate(uploaded_files): holder = target_dir / str(index) holder.mkdir(parents=True, exist_ok=True) path = holder / uploaded.name # Потоковое копирование: UploadedFile уже целиком лежит в памяти # Streamlit, а getvalue() делал бы вторую полную копию каждого # вложения перед записью на диск. uploaded.seek(0) with path.open("wb") as target: shutil.copyfileobj(uploaded, target) paths.append(path) return paths def _render_token_popover() -> None: """Самообслуживание: пользователь сам привязывает свой App Token OpenCloud к своей учётке (по preferred_username из Keycloak), без обращения к администратору с CLI-скриптом. """ with st.popover("Добавить токен"): username = resolve_opencloud_username() if not username: st.error("Не удалось определить вашу учётную запись.") return token = st.text_input( "App Token", type="password", help="Долгоживущий App Token из веб-интерфейса OpenCloud (6 слов через пробел).", key="opencloud_token_input", ) if st.button("Сохранить", key="opencloud_token_save"): if not token: st.error("Введите токен.") else: try: asyncio.run(add_token(username, token, added_by=username)) except OpenCloudTokenStoreError as exc: st.error(f"Не удалось сохранить токен: {exc}") else: st.success("Токен сохранён.") # С новым токеном квоту можно узнать заново (без токена # в кеше лежит None). st.session_state.pop(QUOTA_STATE_KEY, None) if __name__ == "__main__": main()