517 lines
23 KiB
Python
517 lines
23 KiB
Python
from __future__ import annotations
|
||
|
||
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,
|
||
ensure_opencloud_folder,
|
||
next_opencloud_run_number,
|
||
resolve_personal_drive_id,
|
||
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"
|
||
|
||
# Вызывается после каждого обработанного XML-файла: (номер файла, всего
|
||
# файлов, имя файла). Нужен, чтобы вызывающий код (main.py) мог показывать
|
||
# прогресс длинной обработки вместо одного блокирующего спиннера — см.
|
||
# комментарий в run_batch про причину (жалоба СФМ на зависание RAISA при
|
||
# батче 230 файлов).
|
||
ProgressCallback = Callable[[int, int, str], None]
|
||
|
||
|
||
@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)
|
||
duration_seconds: float = 0.0
|
||
# Заполняются, только если файлы обработаны, но загрузка в 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:
|
||
opencloud_space_id = resolve_personal_drive_id(
|
||
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 мин — нелинейно, причина
|
||
# пока не локализована.
|
||
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.
|
||
# opencloud_space_id получен в начале run_batch (проверка
|
||
# доступа до обработки).
|
||
run_number = next_opencloud_run_number(
|
||
opencloud_space_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)
|
||
report_path, protocol_path, log_path = _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,
|
||
space_id=opencloud_space_id,
|
||
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))
|
||
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=report_path,
|
||
protocol_path=protocol_path,
|
||
log_path=log_path,
|
||
messages=messages,
|
||
duration_seconds=duration_seconds,
|
||
)
|
||
|
||
|
||
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,
|
||
space_id: str,
|
||
opencloud_username: str,
|
||
opencloud_access_token: str,
|
||
year: str,
|
||
month: str,
|
||
day: str,
|
||
run_number: int,
|
||
) -> tuple[str, str, str]:
|
||
# space_id (личное пространство пользователя) резолвится вызывающим
|
||
# кодом заново перед каждой загрузкой (не кешируем) — см.
|
||
# resolve_personal_drive_id. Кладём файлы по структуре
|
||
# Out/год/месяц/день/номер_запуска.
|
||
#
|
||
# opencloud_access_token здесь — App Token заказчика (не Bearer/ID-токен),
|
||
# авторизация идёт через HTTP Basic Auth вместе с opencloud_username.
|
||
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_url = 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_url = 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_url = 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,
|
||
)
|
||
return report_url, protocol_url, log_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} сек"
|