RSHB/SFM
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix 10% write in report to 95%

Browse Source
This commit is contained in:
Raykov-MS 2026-07-06 13:27:10 +03:00
parent 636acd1087
commit 27a11ae3ee
6 changed files with 583 additions and 419 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
app/pipeline/mapping.py
View File

@ -85,14 +85,14 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule,
path = str(row.get("xml_path", "")).strip()
if not column or tag in _INVALID_TAGS:
continue
candidates = [tag]
candidates = list(_extract_tag_candidate_keys(tag))
if column.startswith("ИНН"):
candidates = ["ИННФЛ", "ИННЮЛ", "ИНН", *candidates]
candidates = ["ИННФЛИП", "ИННФЛ", "ИННЮЛ", "ИНН", *candidates]
if column == "Тип участника":
candidates = ["Тип", "ТипУчастника", *candidates]
if path:
path_tag = path.strip("/").split("/")[-1]
if path_tag and path_tag not in candidates:
for path_tag in _extract_path_candidate_keys(path):
if path_tag not in candidates:
candidates.append(path_tag)
rules.append(
MappingRule(
@ -108,6 +108,32 @@ def _build_rules(mapping_data: list[dict[str, str | int]]) -> tuple[MappingRule,
return tuple(rules)
def _extract_tag_candidate_keys(tag: str) -> tuple[str, ...]:
candidates: list[str] = []
for raw_tag in re.split(r"[|,]", tag):
cleaned_tag = raw_tag.strip()
if not cleaned_tag or cleaned_tag in _INVALID_TAGS:
continue
if cleaned_tag not in candidates:
candidates.append(cleaned_tag)
return tuple(candidates)
def _extract_path_candidate_keys(path: str) -> tuple[str, ...]:
candidates: list[str] = []
for raw_path in re.findall(r"/[^\s|]+", path):
parts = [part for part in raw_path.strip("/").split("/") if part]
if not parts:
continue
dotted_path = ".".join(parts)
if dotted_path and dotted_path not in candidates:
candidates.append(dotted_path)
leaf_tag = parts[-1]
if leaf_tag and leaf_tag not in _INVALID_TAGS and leaf_tag not in candidates:
candidates.append(leaf_tag)
return tuple(candidates)
_MAPPING_DATA = _load_mapping_data()
FIXED_REPORT_COLUMNS: tuple[str, ...] = _build_fixed_columns(_MAPPING_DATA)
REPORT_MAPPING_RULES: tuple[MappingRule, ...] = _build_rules(_MAPPING_DATA)
@ -154,4 +180,10 @@ def _pick_value(payload: dict[str, str], candidates: tuple[str, ...]) -> str:
short = key.split(".")[-1]
if short in payload and payload[short]:
return payload[short]
dotted_suffix = f".{short}"
for payload_key, payload_value in payload.items():
if not payload_value:
continue
if payload_key.endswith(dotted_suffix):
return payload_value
return ""

532
app/pipeline/mapping_table.json
View File

File diff suppressed because it is too large Load Diff

56
app/pipeline/report.py
View File

@ -5,8 +5,8 @@ from pathlib import Path
from typing import Iterable
from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from .mapping import FIXED_REPORT_COLUMNS, build_fixed_row
@ -20,16 +20,15 @@ class ReportRow:
participant_fields: dict[str, str]
def write_report(rows: Iterable[ReportRow], destination: Path) -> None:
row_list = list(rows)
columns = list(FIXED_REPORT_COLUMNS)
workbook = Workbook()
sheet = workbook.active
sheet.title = "Отчет"
sheet.freeze_panes = "A2"
class StreamingReportWriter:
def __init__(self) -> None:
self.columns = list(FIXED_REPORT_COLUMNS)
self.workbook = Workbook(write_only=True)
self.sheet = self.workbook.create_sheet(title="Отчет")
self.sheet.freeze_panes = "A2"
self.sheet.append(self._build_header_row())
_write_header(sheet, columns)
for row_index, row in enumerate(row_list, start=2):
def append_row(self, row: ReportRow) -> None:
data = build_fixed_row(
file_name=row.file_name,
record_id=row.record_id,
@ -37,33 +36,38 @@ def write_report(rows: Iterable[ReportRow], destination: Path) -> None:
operation_fields=row.operation_fields,
participant_fields=row.participant_fields,
)
for col_index, column_name in enumerate(columns, start=1):
cell = sheet.cell(
row=row_index, column=col_index, value=data.get(column_name, "")
)
cells: list[WriteOnlyCell] = []
for column_name in self.columns:
cell = WriteOnlyCell(self.sheet, value=data.get(column_name, ""))
if "Дата" in column_name:
cell.number_format = "DD.MM.YYYY"
if "Сумм" in column_name or "Сумма" in column_name:
cell.number_format = "#,##0.00"
cells.append(cell)
self.sheet.append(cells)
_auto_width(sheet, columns)
def save(self, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
workbook.save(destination)
self.workbook.save(destination)
def _write_header(sheet, columns: list[str]) -> None:
def _build_header_row(self) -> list[WriteOnlyCell]:
fill = PatternFill(start_color="E2F0D9", end_color="E2F0D9", fill_type="solid")
for index, name in enumerate(columns, start=1):
cell = sheet.cell(row=1, column=index, value=name)
result: list[WriteOnlyCell] = []
for name in self.columns:
cell = WriteOnlyCell(self.sheet, value=name)
cell.font = Font(bold=True)
cell.alignment = Alignment(
horizontal="center", vertical="center", wrap_text=True
horizontal="center",
vertical="center",
wrap_text=True,
)
cell.fill = fill
result.append(cell)
return result
def _auto_width(sheet, columns: list[str]) -> None:
for index, column_name in enumerate(columns, start=1):
width = max(12, min(60, len(column_name) + 4))
column_letter = get_column_letter(index)
sheet.column_dimensions[column_letter].width = width
def write_report(rows: Iterable[ReportRow], destination: Path) -> None:
writer = StreamingReportWriter()
for row in rows:
writer.append_row(row)
writer.save(destination)

223
app/pipeline/service.py
View File

@ -23,7 +23,7 @@ from .file_manager import (
)
from .parser import FileParseResult, parse_xml_content
from .protocol import ProtocolRow, write_protocol
from .report import ReportRow, write_report
from .report import ReportRow, StreamingReportWriter
EMPTY_BATCH_STATUS = "info"
@ -80,7 +80,6 @@ def run_batch(
)
run_context = create_run_context(input_dir=input_dir, samba_conn=samba_conn)
report_rows: list[ReportRow] = []
protocol_rows: list[ProtocolRow] = []
log_lines: list[str] = []
@ -89,14 +88,121 @@ def run_batch(
f"{_now()} | scan files: valid={len(scan_result.valid_files)} invalid={len(scan_result.invalid_files)} "
)
report_file_name = build_report_file_name(
run_context.date_key, run_context.run_number
)
protocol_file_name = build_protocol_file_name(
run_context.date_key, run_context.run_number
)
log_file_name = build_process_log_file_name(
run_context.date_key, run_context.run_number
)
report_path = join_path(run_context.run_dir, report_file_name)
protocol_path = join_path(run_context.run_dir, protocol_file_name)
log_path = join_path(run_context.run_dir, log_file_name)
with tempfile.TemporaryDirectory() as temp_dir:
temp_dir_path = Path(temp_dir)
local_report = temp_dir_path / "report.xlsx"
local_protocol = temp_dir_path / "protocol.xlsx"
local_log = temp_dir_path / "process.log"
report_writer = StreamingReportWriter()
_process_files(
scan_result=scan_result,
run_context=run_context,
samba_conn=samba_conn,
launcher=launcher,
report_writer=report_writer,
protocol_rows=protocol_rows,
log_lines=log_lines,
)
if not protocol_rows:
protocol_rows.append(
ProtocolRow(
file_name="-",
status=EMPTY_BATCH_STATUS,
message="Новых файлов для обработки нет.",
operations_total=0,
rows_written=0,
launcher=launcher,
processed_at=datetime.now(),
)
)
log_lines.append(f"{_now()} | empty batch")
report_writer.save(local_report)
write_protocol(protocol_rows, local_protocol)
# Use CRLF to keep one-record-per-line in Windows viewers.
local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8")
_upload_artifacts(
local_report=local_report,
local_protocol=local_protocol,
local_log=local_log,
report_path=report_path,
protocol_path=protocol_path,
log_path=log_path,
samba_conn=samba_conn,
)
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")
)
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,
)
def _parse_remote_xml(file_name: str, remote_path: str, samba_conn) -> FileParseResult:
try:
content = read_remote_file_bytes(remote_path=remote_path, samba_conn=samba_conn)
except Exception as exc: # noqa: BLE001
result = FileParseResult(file_name=file_name)
result.fatal_error = f"Ошибка чтения файла из SAMBA: {exc}"
return result
return parse_xml_content(file_name=file_name, xml_content=content)
def _process_files(
scan_result,
run_context,
samba_conn,
launcher: str,
report_writer: StreamingReportWriter,
protocol_rows: list[ProtocolRow],
log_lines: list[str],
) -> None:
for remote_file in scan_result.invalid_files:
reason = "Некорректное имя файла"
reason = "Некорректное имя XML-файла"
moved_to = ""
try:
moved_to = move_file_to_error(
file_path=remote_file.path,
file_name=remote_file.name,
error_dir=run_context.error_dir,
samba_conn=samba_conn,
)
except Exception as exc: # noqa: BLE001
reason = f"{reason}. Не удалось переместить в Err: {exc}"
log_lines.append(
f"{_now()} | move to Err failed: {remote_file.path}. {exc}"
)
protocol_rows.append(
ProtocolRow(
file_name=remote_file.name,
@ -115,7 +221,16 @@ def run_batch(
for remote_file in scan_result.valid_files:
parse_result = _parse_remote_xml(remote_file.name, remote_file.path, samba_conn)
report_rows.extend(_report_rows_from_parse(parse_result))
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,
)
)
status, message = _resolve_parse_status(parse_result)
if status == "success":
try:
@ -164,89 +279,6 @@ def run_batch(
)
)
if not protocol_rows:
protocol_rows.append(
ProtocolRow(
file_name="-",
status=EMPTY_BATCH_STATUS,
message="Новых файлов для обработки нет.",
operations_total=0,
rows_written=0,
launcher=launcher,
processed_at=datetime.now(),
)
)
log_lines.append(f"{_now()} | empty batch")
report_file_name = build_report_file_name(
run_context.date_key, run_context.run_number
)
protocol_file_name = build_protocol_file_name(
run_context.date_key, run_context.run_number
)
log_file_name = build_process_log_file_name(
run_context.date_key, run_context.run_number
)
report_path = join_path(run_context.run_dir, report_file_name)
protocol_path = join_path(run_context.run_dir, protocol_file_name)
log_path = join_path(run_context.run_dir, log_file_name)
_upload_artifacts(
report_rows=report_rows,
protocol_rows=protocol_rows,
log_lines=log_lines,
report_path=report_path,
protocol_path=protocol_path,
log_path=log_path,
samba_conn=samba_conn,
)
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")
)
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,
)
def _parse_remote_xml(file_name: str, remote_path: str, samba_conn) -> FileParseResult:
try:
content = read_remote_file_bytes(remote_path=remote_path, samba_conn=samba_conn)
except Exception as exc: # noqa: BLE001
result = FileParseResult(file_name=file_name)
result.fatal_error = f"Ошибка чтения файла из SAMBA: {exc}"
return result
return parse_xml_content(file_name=file_name, xml_content=content)
def _report_rows_from_parse(parse_result: FileParseResult) -> list[ReportRow]:
result: list[ReportRow] = []
for row in parse_result.rows:
result.append(
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,
)
)
return result
def _resolve_parse_status(parse_result: FileParseResult) -> tuple[str, str]:
if parse_result.fatal_error:
@ -261,25 +293,14 @@ def _resolve_parse_status(parse_result: FileParseResult) -> tuple[str, str]:
def _upload_artifacts(
report_rows: list[ReportRow],
protocol_rows: list[ProtocolRow],
log_lines: list[str],
local_report: Path,
local_protocol: Path,
local_log: Path,
report_path: str,
protocol_path: str,
log_path: str,
samba_conn,
) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
temp_dir_path = Path(temp_dir)
local_report = temp_dir_path / "report.xlsx"
local_protocol = temp_dir_path / "protocol.xlsx"
local_log = temp_dir_path / "process.log"
write_report(report_rows, local_report)
write_protocol(protocol_rows, local_protocol)
# Use CRLF to keep one-record-per-line in Windows viewers.
local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8")
upload_local_file(local_report, report_path, samba_conn=samba_conn)
upload_local_file(local_protocol, protocol_path, samba_conn=samba_conn)
upload_local_file(local_log, log_path, samba_conn=samba_conn)

54
tests/unit/test_mapping.py
View File

@ -1,6 +1,10 @@
from __future__ import annotations
from app.pipeline.mapping import FIXED_REPORT_COLUMNS, build_fixed_row
from app.pipeline.mapping import (
FIXED_REPORT_COLUMNS,
_extract_tag_candidate_keys,
build_fixed_row,
)
def _column_with(prefix: str) -> str:
@ -69,3 +73,51 @@ def test_build_fixed_row_does_not_take_operation_inn_for_participant() -> None:
def test_fixed_report_columns_over_200() -> None:
assert len(FIXED_REPORT_COLUMNS) == 203
def test_build_fixed_row_uses_alternative_path_leaf_for_emitent_name() -> None:
row = build_fixed_row(
file_name="f.xml",
record_id="R1",
operation_index=1,
operation_fields={
"ФИОСтрока": "Эмитент ФЛ",
},
participant_fields={},
)
assert row["Наименование/ ФИО эмитента ЦП"] == "Эмитент ФЛ"
def test_build_fixed_row_maps_dotted_participant_keys() -> None:
row = build_fixed_row(
file_name="f.xml",
record_id="R1",
operation_index=1,
operation_fields={},
participant_fields={
"УчастникФЛИП.СведФЛИП.ИННФЛИП": "123456789012",
"УчастникФЛИП.СведФЛИП.СведДокУдЛичн.НомДок": "451099",
},
)
assert row[_column_with("ИНН ")] == "123456789012"
assert row["Номер"] == "451099"
def test_build_fixed_row_splits_pipe_separated_xml_tag_values() -> None:
row = build_fixed_row(
file_name="f.xml",
record_id="R1",
operation_index=1,
operation_fields={
"Индекс": "674500",
},
participant_fields={},
)
assert row["Адрес по структуре (целый)"] == "674500"
def test_extract_tag_candidate_keys_supports_comma_separator() -> None:
candidates = _extract_tag_candidate_keys(
"КППЮЛ, ИдентификацияФЛ, ПризнакОргФормаИНБОЮЛ"
)
assert candidates == ("КППЮЛ", "ИдентификацияФЛ", "ПризнакОргФормаИНБОЮЛ")

55
tests/unit/test_report.py Normal file
View File

@ -0,0 +1,55 @@
from __future__ import annotations
from pathlib import Path
from openpyxl import load_workbook
from app.pipeline.mapping import FIXED_REPORT_COLUMNS
from app.pipeline.report import ReportRow, write_report
def test_write_report_streaming_preserves_header_and_formats(tmp_path: Path) -> None:
destination = tmp_path / "report.xlsx"
rows = [
ReportRow(
file_name="SKO115FZ_01_123456789_20260616_X00001.xml",
record_id="RID-1",
operation_index=1,
operation_fields={"СуммаОпер": "100.50", "ДатаОпер": "2026-06-16"},
participant_fields={},
)
]
write_report(rows, destination)
workbook = load_workbook(destination)
sheet = workbook["Отчет"]
headers = [cell.value for cell in sheet[1]]
assert headers == list(FIXED_REPORT_COLUMNS)
row_values = [
sheet.cell(row=2, column=index).value
for index in range(1, len(FIXED_REPORT_COLUMNS) + 1)
]
assert "SKO115FZ_01_123456789_20260616_X00001.xml" in row_values
date_col_idx = next(
(
index
for index, column in enumerate(FIXED_REPORT_COLUMNS, start=1)
if "Дата" in column
),
None,
)
if date_col_idx is not None:
assert sheet.cell(row=2, column=date_col_idx).number_format == "DD.MM.YYYY"
sum_col_idx = next(
(
index
for index, column in enumerate(FIXED_REPORT_COLUMNS, start=1)
if "Сумм" in column or "Сумма" in column
),
None,
)
if sum_col_idx is not None:
assert sheet.cell(row=2, column=sum_col_idx).number_format == "#,##0.00"