# Copyright (C) 2023-2026 Pain001. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
#
# See the License for the specific language governing permissions and
# limitations under the License.
"""End-to-end orchestration of ISO 20022 payment file generation."""
import logging
import os
import sys
import time
from typing import Any
import pain001.xml.generate_xml as xml_generate
import pain001.xml.register_namespaces as xml_namespaces
from pain001.constants import valid_xml_types
from pain001.context.context import Context
from pain001.data.loader import load_payment_data, load_payment_data_streaming
from pain001.exceptions import XMLGenerationError
from pain001.logging_schema import (
Events,
Fields,
log_event,
log_process_error,
log_process_start,
log_process_success,
)
from pain001.observability import (
emit_metric_event,
set_span_attributes,
traced,
)
from pain001.security.path_validator import sanitize_for_log, validate_path
# Library code: no handlers, no level overrides — the host
# application controls logging configuration.
logger = logging.getLogger(__name__)
def _validate_inputs(
xml_message_type: str,
xml_template_file_path: str,
xsd_schema_file_path: str,
) -> tuple[str, str]:
"""Validate message type and required file paths.
Args:
xml_message_type: ISO 20022 message type (e.g., 'pain.001.001.03').
xml_template_file_path: Path to the XML template file.
xsd_schema_file_path: Path to the XSD schema file.
Returns:
Tuple of (validated template path, validated schema path).
Raises:
XMLGenerationError: If the XML message type is not supported.
FileNotFoundError: If the template or schema file does not exist
or fails path validation.
"""
context_logger = Context.get_instance().get_logger()
if xml_message_type not in valid_xml_types:
error_message = (
f"Error: Invalid XML message type: '{xml_message_type}'."
)
context_logger.error(
f"{sanitize_for_log(error_message)}".replace("\n", "")
)
log_event(
logger,
logging.ERROR,
Events.VALIDATION_ERROR,
**{
Fields.VALIDATION_TYPE: "message_type",
Fields.MESSAGE_TYPE: xml_message_type,
Fields.ERROR_MESSAGE: error_message,
},
)
raise XMLGenerationError(error_message)
try:
safe_template_path = validate_path(
xml_template_file_path, must_exist=True
)
except Exception as e:
error_message = f"Error: XML template '{xml_template_file_path}' does not exist or is invalid: {e}."
context_logger.error(
f"{sanitize_for_log(error_message)}".replace("\n", "")
)
log_event(
logger,
logging.ERROR,
Events.VALIDATION_ERROR,
**{
Fields.VALIDATION_TYPE: "template_file",
Fields.TEMPLATE_PATH: xml_template_file_path,
Fields.ERROR_MESSAGE: error_message,
},
)
raise FileNotFoundError(error_message) from e
try:
safe_schema_path = validate_path(xsd_schema_file_path, must_exist=True)
except Exception as e:
error_message = f"Error: XSD schema file '{xsd_schema_file_path}' does not exist or is invalid: {e}."
context_logger.error(
f"{sanitize_for_log(error_message)}".replace("\n", "")
)
log_event(
logger,
logging.ERROR,
Events.VALIDATION_ERROR,
**{
Fields.VALIDATION_TYPE: "schema_file",
Fields.SCHEMA_PATH: xsd_schema_file_path,
Fields.ERROR_MESSAGE: error_message,
},
)
raise FileNotFoundError(error_message) from e
return str(safe_template_path), str(safe_schema_path)
def _determine_data_source_type(
data_file_path: str | list[dict[str, Any]] | dict[str, Any],
) -> str:
"""Determine the type of the data source."""
if isinstance(data_file_path, list):
return "list"
if isinstance(data_file_path, dict):
return "dict"
if not isinstance(data_file_path, str):
return "unknown"
if data_file_path.endswith(".db") or "sqlite" in data_file_path:
return "sqlite"
for ext in [".csv", ".jsonl", ".json", ".parquet"]:
if data_file_path.endswith(ext):
return ext.lstrip(".")
return "file"
def _load_data(
data_file_path: str | list[dict[str, Any]] | dict[str, Any],
start_time: float,
) -> list[dict[str, Any]]:
"""Load and validate payment data from files or Python objects."""
data_source_kind = _determine_data_source_type(data_file_path)
log_event(
logger,
logging.INFO,
Events.DATA_LOAD_START,
**{Fields.DATA_SOURCE_TYPE: data_source_kind},
)
try:
payment_data = load_payment_data(data_file_path)
duration_ms = int((time.time() - start_time) * 1000)
file_size_bytes = None
if isinstance(data_file_path, str) and os.path.exists(data_file_path):
file_size_bytes = os.path.getsize(data_file_path)
emit_metric_event(
"file_loaded",
data_source_type=data_source_kind,
record_count=len(payment_data),
file_size_bytes=file_size_bytes,
duration_ms=duration_ms,
)
log_event(
logger,
logging.INFO,
Events.DATA_LOAD_SUCCESS,
**{
Fields.DATA_SOURCE_TYPE: data_source_kind,
Fields.RECORD_COUNT: len(payment_data),
Fields.DURATION_MS: duration_ms,
},
)
return payment_data
except (FileNotFoundError, ValueError) as e:
duration_ms = int((time.time() - start_time) * 1000)
log_event(
logger,
logging.ERROR,
Events.DATA_LOAD_ERROR,
**{
Fields.DATA_SOURCE_TYPE: data_source_kind,
Fields.ERROR_TYPE: type(e).__name__,
Fields.ERROR_MESSAGE: str(e),
Fields.DURATION_MS: duration_ms,
},
)
raise
def _register_message_namespaces(xml_message_type: str) -> None:
"""Register XML namespace prefixes and URIs for the given message type."""
log_event(
logger,
logging.INFO,
Events.NAMESPACE_REGISTER,
**{Fields.MESSAGE_TYPE: xml_message_type},
)
xml_namespaces.register_namespaces(xml_message_type)
def _generate_and_log(
payment_data: list[dict[str, Any]],
xml_message_type: str,
xml_template_file_path: str,
xsd_schema_file_path: str,
output_path: str | None = None,
) -> tuple[str, int]:
"""Generate the XML, returning (output file path, duration in ms)."""
gen_start = time.time()
log_event(
logger,
logging.INFO,
Events.XML_GENERATE_START,
**{
Fields.MESSAGE_TYPE: xml_message_type,
Fields.RECORD_COUNT: len(payment_data),
},
)
written_path = xml_generate.generate_xml(
payment_data,
xml_message_type,
xml_template_file_path,
xsd_schema_file_path,
output_path=output_path,
)
return written_path, int((time.time() - gen_start) * 1000)
[docs]
@traced("pain001.generate")
def process_files(
xml_message_type: str,
xml_template_file_path: str,
xsd_schema_file_path: str,
data_file_path: str | list[dict[str, Any]] | dict[str, Any],
output_path: str | None = None,
) -> str:
"""
Generate an ISO 20022 payment message from various data sources.
Args:
xml_message_type: XML message type (e.g., 'pain.001.001.03').
xml_template_file_path: Path to the XML template file.
xsd_schema_file_path: Path to the XSD schema file.
data_file_path: File path (CSV/DB/JSON/Parquet) or Python data (list/dict).
output_path: Explicit path for the generated XML file. When omitted,
the file is written next to the template (deprecated; requires
the template to live under the current working directory).
Returns:
The path the generated XML file was written to.
Raises:
XMLGenerationError: If the message type is not supported or the
XML file could not be written.
Exception: Any error raised while validating inputs, loading
data, or generating XML (e.g. FileNotFoundError for missing
files, PaymentValidationError for bad amounts) is logged and
re-raised unchanged.
"""
context_logger = Context.get_instance().get_logger()
data_source_kind = _determine_data_source_type(data_file_path)
start_time = log_process_start(logger, xml_message_type, data_source_kind)
try:
safe_template_path, safe_schema_path = _validate_inputs(
xml_message_type, xml_template_file_path, xsd_schema_file_path
)
payment_data = _load_data(data_file_path, start_time)
# Stamp the active OTel span (a no-op when OTEL_ENABLED is
# unset) so operators can pivot traces by message type +
# row count.
set_span_attributes(
**{
"pain001.message_type": xml_message_type,
"pain001.row_count": len(payment_data),
}
)
_register_message_namespaces(xml_message_type)
written_path, gen_duration = _generate_and_log(
payment_data,
xml_message_type,
safe_template_path,
safe_schema_path,
output_path=output_path,
)
if os.path.exists(written_path):
context_logger.info(
f"Successfully generated XML file '{written_path}'".replace(
"\n", ""
)
)
log_process_success(
logger,
start_time,
xml_message_type,
len(payment_data),
generation_ms=gen_duration,
)
else:
error_msg = f"Failed to generate XML file at '{written_path}'"
context_logger.error(
f"{sanitize_for_log(error_msg)}".replace("\n", "")
)
log_event(
logger,
logging.ERROR,
Events.XML_GENERATE_ERROR,
**{
Fields.MESSAGE_TYPE: xml_message_type,
Fields.TEMPLATE_PATH: written_path,
Fields.ERROR_MESSAGE: error_msg,
},
)
raise XMLGenerationError(error_msg)
return written_path
except Exception as e:
log_process_error(logger, e, xml_message_type)
raise
[docs]
def process_files_streaming(
xml_message_type: str,
xml_template_file_path: str,
xsd_schema_file_path: str,
data_file_path: str,
chunk_size: int = 1000,
output_dir: str | None = None,
) -> list[str]:
"""Generate multiple XML files from streamed input chunks.
Args:
xml_message_type: XML message type (e.g., 'pain.001.001.03').
xml_template_file_path: Path to the XML template file.
xsd_schema_file_path: Path to the XSD schema file.
data_file_path: Path to the payment data file.
chunk_size: Rows per chunk; each chunk becomes one XML file.
output_dir: Directory to write chunked XML files to. When
omitted, files are written next to the data file.
Returns:
Paths of the generated chunk XML files, in chunk order.
"""
safe_template_path, safe_schema_path = _validate_inputs(
xml_message_type, xml_template_file_path, xsd_schema_file_path
)
_register_message_namespaces(xml_message_type)
if output_dir is None:
output_dir = os.path.dirname(os.path.realpath(data_file_path))
generated_paths: list[str] = []
for chunk_index, payment_chunk in enumerate(
load_payment_data_streaming(
data_file_path, chunk_size=chunk_size, validate=False
),
start=1,
):
xml_content = xml_generate.generate_xml_string(
payment_chunk,
xml_message_type,
safe_template_path,
safe_schema_path,
)
chunked_path = _chunk_output_path(
os.path.join(output_dir, f"{xml_message_type}.xml"), chunk_index
)
with open(chunked_path, "w", encoding="utf-8") as handle:
handle.write(xml_content)
emit_metric_event(
"xml_generated",
message_type=xml_message_type,
output_path=chunked_path,
chunk_index=chunk_index,
record_count=len(payment_chunk),
)
generated_paths.append(chunked_path)
return generated_paths
def _chunk_output_path(base_xml_path: str, chunk_index: int) -> str:
"""Build a chunk-specific XML output path."""
stem, suffix = os.path.splitext(base_xml_path)
return f"{stem}.chunk{chunk_index:04d}{suffix}"
if __name__ == "__main__":
if len(sys.argv) < 5:
print(
"Usage: python3 -m pain001 "
+ " ".join(
[
"<xml_message_type>",
"<xml_template_file_path>",
"<xsd_schema_file_path>",
"<data_file_path>",
]
)
)
sys.exit(1)
process_files(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4])