from __future__ import annotations import tempfile import time import zipfile from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from ..opencloud_access import ( OpenCloudUploadError, ensure_opencloud_folder, resolve_personal_drive_id, upload_local_file_to_opencloud, ) from .config import ( DEFAULT_OUTPUT_DIR_NAME, build_date_parts, build_process_log_file_name, build_protocol_file_name, build_report_file_name, ) from .file_manager import ( ProcessedCopySucceededDeleteFailedError, check_dir_write_access, check_input_dir_access, connect_samba, create_run_context, ensure_dir_exists, move_file_to_error, move_file_to_processed, read_remote_file_bytes, scan_input_files, ) from .parser import FileParseResult, parse_xml_content from .protocol import ProtocolRow, write_protocol from .report import ReportRow, StreamingReportWriter EMPTY_BATCH_STATUS = "info" @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 def run_batch( smb_base_path: str, input_dir: str, samba_user: str, samba_password: str, launcher: str, opencloud_username: str | None, opencloud_access_token: str | None, artifacts_dir: str | None = None, ) -> BatchResult: started_at = time.perf_counter() messages: list[str] = [] if not samba_user: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=["Введите логин SAMBA."], ) 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 для этого пользователя " "(проверьте таблицу токенов)." ], ) samba_conn = connect_samba( smb_base_path=smb_base_path, user_params={"user_name": samba_user, "password": samba_password}, ) try: ensure_dir_exists(input_dir, samba_conn) except Exception as exc: # noqa: BLE001 return _build_input_dir_error_result( input_dir=input_dir, details=str(exc), ) if not samba_conn.path_exists(input_dir): return _build_input_dir_error_result( input_dir=input_dir, details="Проверьте права на родительский каталог.", ) is_ok, access_error = check_input_dir_access( input_dir=input_dir, samba_conn=samba_conn ) if not is_ok: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[access_error or "Ошибка доступа к входной папке."], ) workspace_dir = artifacts_dir or input_dir run_context = create_run_context(input_dir=workspace_dir, samba_conn=samba_conn) checks = [ (input_dir, "Входная папка (перемещение файлов)"), (run_context.run_dir, "Папка Out (текущий запуск)"), (run_context.processed_dir, "Папка Processed"), (run_context.error_dir, "Папка Err"), ] preflight_labels: list[str] = [] for target_dir, label in checks: is_ok, write_error = check_dir_write_access( target_dir=target_dir, samba_conn=samba_conn, source_name=label, ) if not is_ok: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[write_error or f"Нет прав на запись в: {target_dir}"], ) preflight_labels.append(label) messages.append( "Проверка прав (чтение/запись/копирование/удаление) пройдена: " + ", ".join(preflight_labels) + "." ) protocol_rows: list[ProtocolRow] = [] log_lines: list[str] = [] log_lines.append(f"{_now()} | preflight access checks: ok") scan_result = scan_input_files( input_dir=input_dir, samba_conn=samba_conn, processed_dir=run_context.processed_dir, ) log_lines.append( f"{_now()} | scan files: valid={len(scan_result.valid_files)} " f"invalid={len(scan_result.invalid_files)} " f"skipped={len(scan_result.skipped_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 ) 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() stage_timings: dict[str, float] = { "smb_read_seconds": 0.0, "parse_seconds": 0.0, "report_append_seconds": 0.0, "move_seconds": 0.0, } _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, stage_timings=stage_timings, ) 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") log_lines.append( f"{_now()} | stage timings (sec): " f"smb_read={stage_timings['smb_read_seconds']:.3f}, " f"parse={stage_timings['parse_seconds']:.3f}, " f"report_append={stage_timings['report_append_seconds']:.3f}, " f"move={stage_timings['move_seconds']:.3f}" ) 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") try: 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, opencloud_username=opencloud_username, opencloud_access_token=opencloud_access_token, year=run_context.year, month=run_context.month, day=run_context.day, run_number=run_context.run_number, ) except OpenCloudUploadError as exc: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[ *messages, f"Файлы обработаны, но не удалось загрузить результат в " f"OpenCloud: {exc}", ], ) 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.append( _build_result_message( status=status, processed=processed, partial=partial, errors=errors, ) ) 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 run_local_zip_batch( zip_path: Path, launcher: str, opencloud_username: str | None, opencloud_access_token: str | None, ) -> BatchResult: """Локальный тестовый прогон: zip с XML -> отчёт/протокол/лог -> OpenCloud. В отличие от run_batch, НЕ использует SMB вообще — ни для чтения входных файлов, ни для перемещения в Processed/Err. Предназначена для ручного теста всей цепочки (парсинг + загрузка в облако) без сетевой шары. Файлы с некорректным именем/битым XML просто попадают в протокол со статусом "error" — никуда не перемещаются (перемещать их и некуда, это временная распакованная копия). """ 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 для этого пользователя " "(проверьте таблицу токенов)." ], ) protocol_rows: list[ProtocolRow] = [] log_lines: list[str] = [f"{_now()} | local zip batch start: {zip_path.name}"] with tempfile.TemporaryDirectory() as extract_dir: extract_dir_path = Path(extract_dir) try: with zipfile.ZipFile(zip_path) as archive: archive.extractall(extract_dir_path) except zipfile.BadZipFile as exc: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[f"Не удалось распаковать архив: {exc}"], ) xml_paths = sorted(extract_dir_path.rglob("*.xml")) log_lines.append( f"{_now()} | распаковано, найдено XML-файлов: {len(xml_paths)}" ) report_writer = StreamingReportWriter() for xml_path in xml_paths: file_name = xml_path.name try: content = xml_path.read_bytes() except Exception as exc: # noqa: BLE001 protocol_rows.append( ProtocolRow( file_name=file_name, 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}") continue # parse_xml_content сам проверяет имя файла и проставляет # fatal_error, если оно не соответствует шаблону — отдельная # валидация здесь не нужна. parse_result = parse_xml_content(file_name=file_name, xml_content=content) 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, ) ) status, message = _resolve_parse_status(parse_result) protocol_rows.append( ProtocolRow( file_name=file_name, status=status, message=message, operations_total=parse_result.total_operations, rows_written=len(parse_result.rows), launcher=launcher, processed_at=datetime.now(), ) ) log_lines.append(f"{_now()} | parsed {file_name}: {status}") 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") year, month, day, date_key = build_date_parts() run_number = 1 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) 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" report_writer.save(local_report) write_protocol(protocol_rows, local_protocol) local_log.write_text("\r\n".join(log_lines) + "\r\n", encoding="utf-8") try: 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, opencloud_username=opencloud_username, opencloud_access_token=opencloud_access_token, year=year, month=month, day=day, run_number=run_number, ) except OpenCloudUploadError as exc: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[ *log_lines, f"Файлы обработаны, но не удалось загрузить результат " f"в OpenCloud: {exc}", ], ) 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 = [ _build_result_message( status=status, processed=processed, partial=partial, errors=errors ) ] 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 _parse_remote_xml( file_name: str, remote_path: str, samba_conn, stage_timings: dict[str, float] ) -> FileParseResult: try: read_started = time.perf_counter() content = read_remote_file_bytes(remote_path=remote_path, samba_conn=samba_conn) stage_timings["smb_read_seconds"] += time.perf_counter() - read_started except Exception as exc: # noqa: BLE001 result = FileParseResult(file_name=file_name) result.fatal_error = f"Ошибка чтения файла из SAMBA: {exc}" return result parse_started = time.perf_counter() result = parse_xml_content(file_name=file_name, xml_content=content) stage_timings["parse_seconds"] += time.perf_counter() - parse_started return result def _process_files( scan_result, run_context, samba_conn, launcher: str, report_writer: StreamingReportWriter, protocol_rows: list[ProtocolRow], log_lines: list[str], stage_timings: dict[str, float], ) -> None: for remote_file in scan_result.skipped_files: log_lines.append(f"{_now()} | skip already processed: {remote_file.path}") for remote_file in scan_result.invalid_files: reason = "Некорректное имя XML-файла" moved_to = "" try: move_started = time.perf_counter() 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, ) stage_timings["move_seconds"] += time.perf_counter() - move_started except Exception as exc: # noqa: BLE001 stage_timings["move_seconds"] += time.perf_counter() - move_started 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, status="error", message=reason, operations_total=0, rows_written=0, launcher=launcher, processed_at=datetime.now(), ) ) if moved_to: log_lines.append( f"{_now()} | invalid -> err: {remote_file.path} => {moved_to}" ) for remote_file in scan_result.valid_files: parse_result = _parse_remote_xml( remote_file.name, remote_file.path, samba_conn, stage_timings ) for row in parse_result.rows: report_started = time.perf_counter() 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, ) ) stage_timings["report_append_seconds"] += ( time.perf_counter() - report_started ) status, message = _resolve_parse_status(parse_result) if status == "success": move_started = time.perf_counter() try: moved_to = move_file_to_processed( file_path=remote_file.path, file_name=remote_file.name, processed_dir=run_context.processed_dir, samba_conn=samba_conn, ) log_lines.append( f"{_now()} | success -> processed: {remote_file.path} => {moved_to}" ) except ProcessedCopySucceededDeleteFailedError as exc: moved_to = exc.target_path message = ( f"{message} Файл скопирован в Processed, " "но исходник не удален из входной папки из-за ограничений доступа." ) log_lines.append( f"{_now()} | success with delete warning: " f"{remote_file.path} => {moved_to}. {exc}" ) except Exception as exc: # noqa: BLE001 status = "partial" message = f"{message} Не удалось переместить в Processed: {exc}" log_lines.append( f"{_now()} | move to Processed failed: {remote_file.path}. {exc}" ) finally: stage_timings["move_seconds"] += time.perf_counter() - move_started else: move_started = time.perf_counter() 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, ) log_lines.append( f"{_now()} | {status} -> err: {remote_file.path} => {moved_to}" ) except Exception as exc: # noqa: BLE001 message = f"{message} Не удалось переместить в Err: {exc}" log_lines.append( f"{_now()} | move to Err failed: {remote_file.path}. {exc}" ) finally: stage_timings["move_seconds"] += time.perf_counter() - move_started protocol_rows.append( ProtocolRow( file_name=remote_file.name, status=status, message=message, operations_total=parse_result.total_operations, rows_written=len(parse_result.rows), launcher=launcher, processed_at=datetime.now(), ) ) 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, opencloud_username: str, opencloud_access_token: str, year: str, month: str, day: str, run_number: int, ) -> tuple[str, str, str]: # Личное пространство ищем заново перед каждой загрузкой (не кешируем # ID) — см. resolve_personal_drive_id. Кладём файлы по той же структуре # Out/год/месяц/день/номер_запуска, что и на SMB (RunContext.run_dir). # # opencloud_access_token здесь — App Token заказчика (не Bearer/ID-токен), # авторизация идёт через HTTP Basic Auth вместе с opencloud_username. space_id = resolve_personal_drive_id(opencloud_username, opencloud_access_token) 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 _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} сек" def _build_input_dir_error_result(input_dir: str, details: str) -> BatchResult: return BatchResult( status="error", processed=0, partial=0, errors=1, report_path="", protocol_path="", log_path="", messages=[f"Не удалось создать входную папку: {input_dir}. {details}"], )