721712d86a
The uploader took a single .zip, so a day split across several archives, or a handful of loose XML files, needed one run each. - main.py and scripts/local_zip_test_app.py now use accept_multiple_files and take .xml alongside .zip and the tar family. Each attachment is written to its own temp subdirectory, because two attachments may share a name. - run_batch takes input_paths instead of zip_path. - New app/pipeline/attachments.py flattens the attachments into a list of XML files. Archive format is detected from the content rather than the suffix, since exports arrive with the wrong extension often enough. .rar and .7z are recognised only to explain why they are refused. - A broken attachment no longer aborts the run. It becomes one protocol row and the remaining attachments are still processed; the reason is also shown in the UI, not only in the protocol file. - Files repeated across attachments are processed once and the repeats are recorded with status "skip". Attaching the same archive twice would otherwise double every row of the report. - Archives are unpacked member by member with the target path checked against the extraction directory. tarfile.data_filter only exists from Python 3.11.4 and extractall would otherwise let a "../" member escape. - The protocol gains a "Вложение" column, so it is clear which attachment a file came from when names repeat across archives. Checked end to end on the 03.02.2023 export: six attachments (two archives, two loose XML files, one broken archive, one duplicate archive) give 57 files processed, 9 repeats skipped, one attachment refused with its reason, and the report is still produced.
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
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
|
||
# Имя вложения, из которого получен файл. Пусто, если файл прикреплён
|
||
# отдельным .xml — тогда имя вложения совпадает с именем файла.
|
||
source: str = ""
|
||
|
||
|
||
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.source or 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
|