from __future__ import annotations import logging import tempfile import time from collections.abc import Callable, Sequence from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from ..opencloud_access import ( OpenCloudUploadError, OpenCloudUploadResult, PersonalDrive, build_opencloud_folder_url, build_opencloud_web_folder_url, ensure_opencloud_folder, next_opencloud_run_number, resolve_opencloud_file_id, resolve_personal_drive, upload_local_file_to_opencloud, ) from .attachments import collect_xml_attachments from .config import ( DEFAULT_OUTPUT_DIR_NAME, build_date_parts, build_process_log_file_name, build_protocol_file_name, build_report_file_name, ) from .parser import FileParseResult, parse_xml_content from .protocol import ProtocolRow, write_protocol from .report import ReportRow, StreamingReportWriter EMPTY_BATCH_STATUS = "info" logger = logging.getLogger(__name__) # Вызывается после каждого обработанного XML-файла: (номер файла, всего # файлов, имя файла). Нужен, чтобы вызывающий код (main.py) мог показывать # прогресс длинной обработки вместо одного блокирующего спиннера — см. # комментарий в run_batch про причину (жалоба СФМ на зависание RAISA при # батче 230 файлов). ProgressCallback = Callable[[int, int, str], None] @dataclass class BatchResult: status: str processed: int partial: int errors: int # WebDAV-адреса загруженных файлов (/dav/spaces/...) — для протокола и # отладки, пользователю их не показываем. report_path: str protocol_path: str log_path: str messages: list[str] = field(default_factory=list) duration_seconds: float = 0.0 # Ссылки для пользователя: открывают файл (папку запуска) в веб-интерфейсе # OpenCloud. Если ссылку на файл получить не удалось — ссылка на папку # запуска (см. _pick_link); она собирается по пути и есть всегда. report_url: str = "" protocol_url: str = "" log_url: str = "" folder_url: str = "" # Заполняются, только если файлы обработаны, но загрузка в OpenCloud не # удалась: раньше в этом случае отчёт удалялся вместе с временной # папкой, и 40 минут обработки большого батча пропадали впустую. # main.py отдаёт их пользователю через st.download_button. fallback_files: dict[str, bytes] = field(default_factory=dict) def run_batch( input_paths: Sequence[Path], launcher: str, opencloud_username: str | None, opencloud_access_token: str | None, progress_callback: ProgressCallback | None = None, ) -> BatchResult: """Прикреплённые XML и архивы -> отчёт/протокол/лог -> OpenCloud. Входные файлы приходят как есть — через вложения (attach) в интерфейсе (Streamlit file_uploader), без SMB и без выделенной входной папки в OpenCloud. Вложений может быть несколько, и это могут быть как отдельные .xml, так и архивы, вперемешку: их разворачивает в плоский список collect_xml_attachments. Вложение, которое не удалось разобрать, попадает в протокол отдельной строкой и не отменяет обработку остальных. Результат грузится в личное пространство пользователя (Out/год/месяц/день/номер_запуска). Файлы с некорректным именем или битым XML просто попадают в протокол со статусом "error" — никуда не перемещаются (перемещать их и некуда, это временная распакованная копия вложения). progress_callback, если передан, вызывается после разбора каждого XML-файла с (обработано_файлов, всего_файлов, имя_файла). Весь батч по-прежнему выполняется синхронно в одном вызове (отдельный воркер/очередь — отдельная задача), но колбэк даёт вызывающему коду возможность обновлять прогресс по ходу обработки, а не показывать один спиннер на всё время — именно отсутствие видимого прогресса на большом батче (230 файлов, тестирование СФМ) навело на мысль, что приложение "зависло". """ started_at = time.perf_counter() if not opencloud_username: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=["Не удалось определить username для авторизации в OpenCloud."], ) if not opencloud_access_token: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[ "Не найден App Token OpenCloud для вашей учётки. " "Добавьте его через кнопку «Добавить токен»." ], ) if not input_paths: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=["Не прикреплено ни одного файла."], ) # Доступ к OpenCloud проверяем ДО обработки, а не после: на батче из 230 # файлов СФМ получили HTTP 401 от OpenCloud только через ~40 минут # обработки, когда дело дошло до загрузки. Неверный/отозванный App Token # или несовпадение username должны всплывать сразу. try: personal_drive = resolve_personal_drive( opencloud_username, opencloud_access_token ) except OpenCloudUploadError as exc: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[_describe_opencloud_access_error(exc, opencloud_username)], ) protocol_rows: list[ProtocolRow] = [] attachment_failures: list[str] = [] attachment_names = ", ".join(path.name for path in input_paths) log_lines: list[str] = [f"{_now()} | batch start: {attachment_names}"] with tempfile.TemporaryDirectory() as extract_dir: extract_dir_path = Path(extract_dir) stage_started = time.perf_counter() collected = collect_xml_attachments(input_paths, extract_dir_path) extract_seconds = time.perf_counter() - stage_started log_lines.append( f"{_now()} | вложений: {len(input_paths)} " f"(архивов: {collected.archives}, " f"отдельных XML: {collected.loose_xml_files}), " f"найдено XML-файлов: {len(collected.files)}, " f"распаковка: {extract_seconds:.1f} с" ) for problem in collected.problems: protocol_rows.append( ProtocolRow( file_name=problem.source, source=problem.source, status="error", message=f"Вложение не обработано: {problem.reason}", operations_total=0, rows_written=0, launcher=launcher, processed_at=datetime.now(), ) ) attachment_failures.append(f"{problem.source} — {problem.reason}") log_lines.append( f"{_now()} | attachment failed: {problem.source}. {problem.reason}" ) for duplicate in collected.duplicates: protocol_rows.append( ProtocolRow( file_name=duplicate.file_name or duplicate.source, source=duplicate.source, status="skip", message=duplicate.reason, operations_total=0, rows_written=0, launcher=launcher, processed_at=datetime.now(), ) ) log_lines.append( f"{_now()} | duplicate skipped: {duplicate.file_name}. " f"{duplicate.reason}" ) report_writer = StreamingReportWriter() total_files = len(collected.files) total_rows_written = 0 parse_total_seconds = 0.0 write_total_seconds = 0.0 files_loop_started = time.perf_counter() for file_number, collected_xml in enumerate(collected.files, start=1): file_name = collected_xml.file_name try: content = collected_xml.path.read_bytes() except Exception as exc: # noqa: BLE001 protocol_rows.append( ProtocolRow( file_name=file_name, source=collected_xml.source, status="error", message=f"Ошибка чтения файла из вложения: {exc}", operations_total=0, rows_written=0, launcher=launcher, processed_at=datetime.now(), ) ) log_lines.append(f"{_now()} | read failed: {file_name}. {exc}") if progress_callback is not None: progress_callback(file_number, total_files, file_name) continue # parse_xml_content сам проверяет имя файла и проставляет # fatal_error, если оно не соответствует шаблону — отдельная # валидация здесь не нужна. parse_started = time.perf_counter() parse_result = parse_xml_content(file_name=file_name, xml_content=content) parse_seconds = time.perf_counter() - parse_started write_started = time.perf_counter() for row in parse_result.rows: report_writer.append_row( 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, is_continuation=row.is_continuation, ) ) write_seconds = time.perf_counter() - write_started parse_total_seconds += parse_seconds write_total_seconds += write_seconds total_rows_written += len(parse_result.rows) status, message = _resolve_parse_status(parse_result) protocol_rows.append( ProtocolRow( file_name=file_name, source=collected_xml.source, status=status, message=message, operations_total=parse_result.total_operations, rows_written=len(parse_result.rows), launcher=launcher, processed_at=datetime.now(), ) ) # Замер на каждый файл — чтобы по логу было видно, растёт ли # время на файл по ходу батча (накопление) или его съедает # конкретный тяжёлый файл (много операций/участников). # Наблюдавшиеся 100 файлов = 6 мин и 230 файлов = 45 мин # объясняются не накоплением: время линейно по числу строк # отчёта (у батчей было разное число операций на файл), а под # работал на 0.25 CPU — профиль «Малый» в raisa_config.yml. log_lines.append( f"{_now()} | parsed {file_name}: {status} | " f"{file_number}/{total_files} | " f"размер={len(content) // 1024} КБ, " f"операций={parse_result.total_operations}, " f"строк={len(parse_result.rows)}, " f"разбор={parse_seconds:.2f} с, " f"запись={write_seconds:.2f} с, " f"всего строк в отчёте={total_rows_written}" ) if progress_callback is not None: progress_callback(file_number, total_files, file_name) if not protocol_rows: protocol_rows.append( ProtocolRow( file_name="-", status=EMPTY_BATCH_STATUS, message="Во вложениях не найдено XML-файлов.", operations_total=0, rows_written=0, launcher=launcher, processed_at=datetime.now(), ) ) log_lines.append(f"{_now()} | empty batch") files_loop_seconds = time.perf_counter() - files_loop_started log_lines.append( f"{_now()} | обработка файлов: {files_loop_seconds:.1f} с " f"(разбор XML: {parse_total_seconds:.1f} с, " f"запись строк отчёта: {write_total_seconds:.1f} с), " f"строк в отчёте: {total_rows_written}" ) year, month, day, date_key = build_date_parts() with tempfile.TemporaryDirectory() as artifacts_dir: artifacts_dir_path = Path(artifacts_dir) local_report = artifacts_dir_path / "report.xlsx" local_protocol = artifacts_dir_path / "protocol.xlsx" local_log = artifacts_dir_path / "process.log" stage_started = time.perf_counter() report_writer.save(local_report) report_save_seconds = time.perf_counter() - stage_started stage_started = time.perf_counter() write_protocol(protocol_rows, local_protocol) protocol_seconds = time.perf_counter() - stage_started log_lines.append( f"{_now()} | сохранение отчёта: {report_save_seconds:.1f} с " f"({local_report.stat().st_size // 1024} КБ), " f"протокол: {protocol_seconds:.1f} с" ) local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8") try: # Номер запуска для папки дня в OpenCloud определяем по # самому OpenCloud — см. next_opencloud_run_number. # personal_drive получен в начале run_batch (проверка # доступа до обработки). run_number = next_opencloud_run_number( personal_drive.id, (DEFAULT_OUTPUT_DIR_NAME, year, month, day), username=opencloud_username, token=opencloud_access_token, ) report_file_name = build_report_file_name(date_key, run_number) protocol_file_name = build_protocol_file_name(date_key, run_number) log_file_name = build_process_log_file_name(date_key, run_number) uploaded = _upload_artifacts( local_report=local_report, local_protocol=local_protocol, local_log=local_log, report_file_name=report_file_name, protocol_file_name=protocol_file_name, log_file_name=log_file_name, personal_drive=personal_drive, opencloud_username=opencloud_username, opencloud_access_token=opencloud_access_token, year=year, month=month, day=day, run_number=run_number, ) except OpenCloudUploadError as exc: # Файлы ещё лежат во временной папке — забираем их в память, # чтобы пользователь мог скачать результат из интерфейса. fallback_name_suffix = date_key fallback_files = { f"report_{fallback_name_suffix}.xlsx": local_report.read_bytes(), f"protocol_{fallback_name_suffix}.xlsx": local_protocol.read_bytes(), f"process_{fallback_name_suffix}.log": local_log.read_bytes(), } return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[ f"Файлы обработаны, но не удалось загрузить результат " f"в OpenCloud: {_describe_opencloud_access_error(exc, opencloud_username)} " f"Отчёт можно скачать кнопками ниже.", *log_lines[-5:], ], duration_seconds=round(time.perf_counter() - started_at, 2), fallback_files=fallback_files, ) 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") ) skipped = sum(1 for row in protocol_rows if row.status == "skip") messages = [ _build_result_message( status=status, processed=processed, partial=partial, errors=errors ) ] if skipped: messages.append(f"Пропущено повторно прикреплённых файлов: {skipped}.") if attachment_failures: messages.append("Не обработаны вложения: " + "; ".join(attachment_failures)) if uploaded.has_fallback_links: messages.append( "Не удалось получить ссылку на файл в веб-интерфейсе OpenCloud — " "такая кнопка откроет папку с результатами." ) duration_seconds = round(time.perf_counter() - started_at, 2) messages.append(f"Время выполнения: {_format_duration(duration_seconds)}.") messages.extend(log_lines[-5:]) return BatchResult( status=status, processed=processed, partial=partial, errors=errors, report_path=uploaded.report.dav_url, protocol_path=uploaded.protocol.dav_url, log_path=uploaded.log.dav_url, messages=messages, duration_seconds=duration_seconds, report_url=uploaded.report_url, protocol_url=uploaded.protocol_url, log_url=uploaded.log_url, folder_url=uploaded.folder_url, ) 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( local_report: Path, local_protocol: Path, local_log: Path, report_file_name: str, protocol_file_name: str, log_file_name: str, personal_drive: PersonalDrive, opencloud_username: str, opencloud_access_token: str, year: str, month: str, day: str, run_number: int, ) -> _UploadedArtifacts: # personal_drive (личное пространство пользователя) резолвится вызывающим # кодом заново перед каждой загрузкой (не кешируем) — см. # resolve_personal_drive. Кладём файлы по структуре # Out/год/месяц/день/номер_запуска. # # opencloud_access_token здесь — App Token заказчика (не Bearer/ID-токен), # авторизация идёт через HTTP Basic Auth вместе с opencloud_username. space_id = personal_drive.id folder_segments = (DEFAULT_OUTPUT_DIR_NAME, year, month, day, str(run_number)) ensure_opencloud_folder( space_id, folder_segments, username=opencloud_username, token=opencloud_access_token, ) report = upload_local_file_to_opencloud( local_report, space_id=space_id, folder_segments=folder_segments, file_name=report_file_name, username=opencloud_username, token=opencloud_access_token, ) protocol = upload_local_file_to_opencloud( local_protocol, space_id=space_id, folder_segments=folder_segments, file_name=protocol_file_name, username=opencloud_username, token=opencloud_access_token, ) log = upload_local_file_to_opencloud( local_log, space_id=space_id, folder_segments=folder_segments, file_name=log_file_name, username=opencloud_username, token=opencloud_access_token, ) # Ссылка на папку запуска собирается по пути и есть всегда; fileId папки # только уточняет её, как в адресной строке самого интерфейса. Если # ссылку на файл получить не удалось, кнопка файла ведёт в эту папку — # обработка из-за ссылок не падает. folder_file_id = resolve_opencloud_file_id( build_opencloud_folder_url(space_id, folder_segments), opencloud_username, opencloud_access_token, ) folder_url = build_opencloud_web_folder_url( _drive_alias(personal_drive, opencloud_username), folder_segments, folder_file_id, ) return _UploadedArtifacts( report=report, protocol=protocol, log=log, report_url=_pick_link("отчёт", report, folder_url), protocol_url=_pick_link("протокол", protocol, folder_url), log_url=_pick_link("лог", log, folder_url), folder_url=folder_url, has_fallback_links=not all(item.web_url for item in (report, protocol, log)), ) def _drive_alias(personal_drive: PersonalDrive, username: str) -> str: if personal_drive.alias: return personal_drive.alias # Так выглядит путь к личному пространству в адресной строке интерфейса: # /files/spaces/personal/<логин>/... logger.warning( "Graph API не вернул driveAlias личного пространства, собираю путь " "из логина: personal/%s", username, ) return f"personal/{username}" @dataclass(frozen=True) class _UploadedArtifacts: report: OpenCloudUploadResult protocol: OpenCloudUploadResult log: OpenCloudUploadResult report_url: str protocol_url: str log_url: str folder_url: str # Хотя бы для одного файла вместо ссылки на него отдана запасная. has_fallback_links: bool def _pick_link(label: str, uploaded: OpenCloudUploadResult, folder_url: str) -> str: if uploaded.web_url: return uploaded.web_url logger.warning( "Нет ссылки на %s (%s) в веб-интерфейсе OpenCloud, показываю " "ссылку на папку запуска", label, uploaded.dav_url, ) return folder_url def _describe_opencloud_access_error(exc: Exception, username: str) -> str: text = str(exc) if "HTTP 401" in text: return ( f"OpenCloud отклонил App Token для пользователя «{username}» (HTTP 401). " "Вероятно, токен истёк, отозван, введён с ошибкой или выпущен под " "другой учётной записью. Создайте новый App Token в OpenCloud и " "сохраните его через «Добавить токен»." ) return f"Нет доступа к OpenCloud: {text}" def _now() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _build_result_message( *, status: str, processed: int, partial: int, errors: int ) -> str: if status == "success": if processed > 0: return f"Все успешно: обработано файлов {processed}." return "Успешно: новых файлов для обработки нет." if status == "partial": return ( "Частично успешно: " f"успешно={processed}, частично={partial}, с ошибками={errors}." ) return f"Ошибка обработки: успешно={processed}, частично={partial}, с ошибками={errors}." def _format_duration(duration_seconds: float) -> str: total_seconds = max(0, int(round(duration_seconds))) minutes, seconds = divmod(total_seconds, 60) if minutes: return f"{minutes} мин {seconds} сек" return f"{seconds} сек"