Seu Template de dados de processamento de pagamentos

from __future__ import annotations

import csv
import hashlib
import re
import sys
import xml.etree.ElementTree as ET
from collections import defaultdict
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple

class=class="hl-string">"hl-comment"># Replace only these documented placeholders.
INPUT_DIRECTORY = Path(class="hl-string">"[Your XML archive directory]")
OUTPUT_FILE = Path(class="hl-string">"[Your ProcessMind event log output path].csv")
START_DATE = class="hl-string">"[YYYY-MM-DD]"
END_DATE = class="hl-string">"[YYYY-MM-DD]"
SOURCE_TIMEZONE = timezone.utc
COMPANY_FILTER = class="hl-string">""  class=class="hl-string">"hl-comment"># Leave empty unless a validated company or business-unit value is required.

class=class="hl-string">"hl-comment"># Update these candidates after inspecting representative FIS XML files.
IDENTIFIER_PATHS = [
    class="hl-string">"EndToEndId", class="hl-string">"TxId", class="hl-string">"InstrId", class="hl-string">"UETR", class="hl-string">"PaymentTransactionId",
    class="hl-string">"PaymentId", class="hl-string">"TransactionId"
]
AMOUNT_PATHS = [class="hl-string">"InstdAmt", class="hl-string">"IntrBkSttlmAmt", class="hl-string">"Amt", class="hl-string">"Amount"]
CURRENCY_PATHS = [class="hl-string">"InstdAmt", class="hl-string">"IntrBkSttlmAmt", class="hl-string">"Amt", class="hl-string">"Currency", class="hl-string">"Ccy"]
DUE_DATE_PATHS = [class="hl-string">"ReqdExctnDt", class="hl-string">"ReqdColltnDt", class="hl-string">"DueDate", class="hl-string">"RequestedExecutionDate"]
SETTLEMENT_DATE_PATHS = [class="hl-string">"SttlmDt", class="hl-string">"SettlementDate", class="hl-string">"ActualSettlementDate"]
TIMESTAMP_PATHS = [
    class="hl-string">"CreDtTm", class="hl-string">"AccptncDtTm", class="hl-string">"AccptncDt", class="hl-string">"OrgnlCreDtTm", class="hl-string">"TxDtTm",
    class="hl-string">"SttlmDt", class="hl-string">"SettlementDate", class="hl-string">"EventTimestamp", class="hl-string">"Timestamp"
]
USER_PATHS = [class="hl-string">"UserId", class="hl-string">"UserID", class="hl-string">"Operator", class="hl-string">"Approver", class="hl-string">"ProcessingUser", class="hl-string">"Initiator"]
PAYMENT_TYPE_PATHS = [class="hl-string">"PmtMtd", class="hl-string">"SvcLvl", class="hl-string">"LclInstrm", class="hl-string">"CtgyPurp", class="hl-string">"PaymentType"]
BUSINESS_UNIT_PATHS = [class="hl-string">"BusinessUnit", class="hl-string">"CompanyCode", class="hl-string">"Company", class="hl-string">"OrgUnit", class="hl-string">"LegalEntity"]
STATUS_PATHS = [class="hl-string">"TxSts", class="hl-string">"GrpSts", class="hl-string">"Status", class="hl-string">"PaymentStatus", class="hl-string">"TransactionStatus"]
REASON_PATHS = [class="hl-string">"Cd", class="hl-string">"Rsn", class="hl-string">"ReasonCode", class="hl-string">"StatusReason", class="hl-string">"AddtlInf"]
ERROR_PATHS = [class="hl-string">"Error", class="hl-string">"Exception", class="hl-string">"Failure", class="hl-string">"ErrorCode", class="hl-string">"ExceptionCode"]
REPAIR_PATHS = [class="hl-string">"Repair", class="hl-string">"Resolved", class="hl-string">"Resolution", class="hl-string">"ErrorResolved", class="hl-string">"ExceptionResolved"]
RECONCILIATION_PATHS = [class="hl-string">"Reconciled", class="hl-string">"ReconciliationStatus", class="hl-string">"LedgerMatched", class="hl-string">"BankStatementMatched"]

class=class="hl-string">"hl-comment"># Validate and extend these mappings against the FIS implementation.
STATUS_ACTIVITY_MAP = {
    class="hl-string">"NEW": class="hl-string">"Payment Request Created",
    class="hl-string">"RECEIVED": class="hl-string">"Payment Request Created",
    class="hl-string">"VALIDATED": class="hl-string">"Payment Details Validated",
    class="hl-string">"PENDING_APPROVAL": class="hl-string">"Payment Sent For Approval",
    class="hl-string">"AWAITING_APPROVAL": class="hl-string">"Payment Sent For Approval",
    class="hl-string">"APPROVAL_PENDING": class="hl-string">"Payment Sent For Approval",
    class="hl-string">"APPROVED": class="hl-string">"Payment Approved",
    class="hl-string">"REJECTED": class="hl-string">"Payment Rejected",
    class="hl-string">"AUTHORIZED": class="hl-string">"Payment Authorized",
    class="hl-string">"SENT": class="hl-string">"Payment Instruction Sent",
    class="hl-string">"DISPATCHED": class="hl-string">"Payment Instruction Sent",
    class="hl-string">"ACCP": class="hl-string">"Payment Confirmed",
    class="hl-string">"ACCEPTED": class="hl-string">"Payment Confirmed",
    class="hl-string">"CONFIRMED": class="hl-string">"Payment Confirmed",
    class="hl-string">"SETTLED": class="hl-string">"Payment Settled",
    class="hl-string">"BOOK": class="hl-string">"Payment Settled",
    class="hl-string">"RECONCILED": class="hl-string">"Payment Reconciled",
}
ERROR_STATUS_CODES = {class="hl-string">"RJCT", class="hl-string">"REJECTED", class="hl-string">"FAILED", class="hl-string">"ERROR", class="hl-string">"EXCEPTION", class="hl-string">"CANCELLED"}
RESOLVED_STATUS_CODES = {class="hl-string">"REPAIRED", class="hl-string">"RESOLVED", class="hl-string">"CORRECTED", class="hl-string">"REPROCESSING", class="hl-string">"REPROCESSED"}

OUTPUT_COLUMNS = [
    class="hl-string">"PaymentTransactionId", class="hl-string">"ActivityName", class="hl-string">"EventTimestamp", class="hl-string">"ProcessingUser",
    class="hl-string">"PaymentAmount", class="hl-string">"CurrencyCode", class="hl-string">"PaymentType", class="hl-string">"PaymentDueDate",
    class="hl-string">"ActualSettlementDate", class="hl-string">"BusinessUnit"
]


def local_name(tag: str) -> str:
    return tag.rsplit(class="hl-string">"}", 1)[-1].lower()


def text_values(root: ET.Element, candidates: Iterable[str]) -> List[str]:
    wanted = {candidate.lower() for candidate in candidates}
    values: List[str] = []
    for element in root.iter():
        if local_name(element.tag) in wanted:
            value = (element.text or class="hl-string">"").strip()
            if value:
                values.append(value)
            for key, attribute_value in element.attrib.items():
                if key.rsplit(class="hl-string">"}", 1)[-1].lower() in {class="hl-string">"ccy", class="hl-string">"currency"} and attribute_value.strip():
                    values.append(attribute_value.strip())
    return values


def first_value(root: ET.Element, candidates: Iterable[str]) -> str:
    values = text_values(root, candidates)
    return values[0] if values else class="hl-string">""


def normalize_code(value: str) -> str:
    return re.sub(rclass="hl-string">"[^A-Z0-9_]", class="hl-string">"", value.upper().replace(class="hl-string">"-", class="hl-string">"_"))


def parse_datetime(value: str) -> Optional[datetime]:
    if not value:
        return None
    candidate = value.strip()
    if candidate.endswith(class="hl-string">"Z"):
        candidate = candidate[:-1] + class="hl-string">"+00:00"
    try:
        parsed = datetime.fromisoformat(candidate)
    except ValueError:
        for fmt in (class="hl-string">"%Y-%m-%d %H:%M:%S", class="hl-string">"%Y-%m-%d", class="hl-string">"%Y%m%d%H%M%S", class="hl-string">"%Y%m%d"):
            try:
                parsed = datetime.strptime(value.strip(), fmt)
                break
            except ValueError:
                parsed = None
        if parsed is None:
            return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=SOURCE_TIMEZONE)
    return parsed.astimezone(timezone.utc)


def output_timestamp(value: Optional[datetime]) -> str:
    return value.astimezone(timezone.utc).isoformat().replace(class="hl-string">"+00:00", class="hl-string">"Z") if value else class="hl-string">""


def output_date(value: str) -> str:
    parsed = parse_datetime(value)
    return parsed.date().isoformat() if parsed else (value[:10] if re.match(rclass="hl-string">"^\d{4}-\d{2}-\d{2}", value) else class="hl-string">"")


def normalize_amount(value: str) -> str:
    if not value:
        return class="hl-string">""
    try:
        return format(Decimal(value.replace(class="hl-string">",", class="hl-string">"").strip()), class="hl-string">"f")
    except InvalidOperation:
        return value.strip()


def source_message_id(path: Path, root: ET.Element) -> str:
    raw = path.as_posix() + class="hl-string">"|" + first_value(root, [class="hl-string">"MsgId", class="hl-string">"MessageId", class="hl-string">"BizMsgIdr"])
    return hashlib.sha256(raw.encode(class="hl-string">"utf-8")).hexdigest()


def in_date_range(event_time: Optional[datetime]) -> bool:
    if not event_time:
        return True
    start = datetime.fromisoformat(START_DATE).replace(tzinfo=timezone.utc)
    end = datetime.fromisoformat(END_DATE).replace(tzinfo=timezone.utc)
    return start <= event_time < end.replace(hour=23, minute=59, second=59, microsecond=999999)


def add_event(events: List[Dict[str, str]], seen: set, transaction_id: str, activity: str,
              event_time: Optional[datetime], user: str, amount: str, currency: str,
              payment_type: str, due_date: str, settlement_date: str, business_unit: str,
              message_id: str) -> None:
    if not transaction_id or not event_time or not in_date_range(event_time):
        return
    row = {
        class="hl-string">"PaymentTransactionId": transaction_id,
        class="hl-string">"ActivityName": activity,
        class="hl-string">"EventTimestamp": output_timestamp(event_time),
        class="hl-string">"ProcessingUser": user,
        class="hl-string">"PaymentAmount": amount,
        class="hl-string">"CurrencyCode": currency,
        class="hl-string">"PaymentType": payment_type,
        class="hl-string">"PaymentDueDate": output_date(due_date),
        class="hl-string">"ActualSettlementDate": output_date(settlement_date),
        class="hl-string">"BusinessUnit": business_unit,
    }
    key = (transaction_id, activity, row[class="hl-string">"EventTimestamp"], message_id)
    if key not in seen:
        seen.add(key)
        events.append(row)


def parse_file(path: Path, events: List[Dict[str, str]], seen: set) -> None:
    try:
        root = ET.parse(path).getroot()
    except (ET.ParseError, OSError) as exc:
        print(fclass="hl-string">"Could not parse {path}: {exc}", file=sys.stderr)
        return
    message_id = source_message_id(path, root)
    transaction_id = first_value(root, IDENTIFIER_PATHS)
    if not transaction_id:
        return
    business_unit = first_value(root, BUSINESS_UNIT_PATHS)
    if COMPANY_FILTER and business_unit != COMPANY_FILTER:
        return
    amount = normalize_amount(first_value(root, AMOUNT_PATHS))
    currency = first_value(root, CURRENCY_PATHS)
    if len(currency) != 3 and currency:
        currency = currency.upper()
    due_date = first_value(root, DUE_DATE_PATHS)
    settlement_date = first_value(root, SETTLEMENT_DATE_PATHS)
    user = first_value(root, USER_PATHS)
    payment_type = first_value(root, PAYMENT_TYPE_PATHS)
    status_values = text_values(root, STATUS_PATHS)
    reason_values = text_values(root, REASON_PATHS)
    error_values = text_values(root, ERROR_PATHS)
    repair_values = text_values(root, REPAIR_PATHS)
    reconciliation_values = text_values(root, RECONCILIATION_PATHS)
    event_time = parse_datetime(first_value(root, TIMESTAMP_PATHS))
    if not event_time:
        return

    message_name = next((local_name(element.tag) for element in root.iter() if local_name(element.tag) in {class="hl-string">"pacs.008", class="hl-string">"pacs.002", class="hl-string">"document"}), class="hl-string">"")
    normalized_statuses = [normalize_code(value) for value in status_values + reason_values]
    normalized_text = class="hl-string">" ".join(normalize_code(value) for value in error_values)
    normalized_repairs = class="hl-string">" ".join(normalize_code(value) for value in repair_values)
    normalized_reconciliation = class="hl-string">" ".join(normalize_code(value) for value in reconciliation_values)

    if message_name or status_values:
        add_event(events, seen, transaction_id, class="hl-string">"Payment Request Created", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if any(code in {class="hl-string">"NEW", class="hl-string">"RECEIVED"} for code in normalized_statuses) else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Details Validated", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if class="hl-string">"VALIDATED" in normalized_statuses else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Sent For Approval", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if any(code in {class="hl-string">"PENDING_APPROVAL", class="hl-string">"AWAITING_APPROVAL", class="hl-string">"APPROVAL_PENDING"} for code in normalized_statuses) else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Approved", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if class="hl-string">"APPROVED" in normalized_statuses else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Rejected", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if any(code in {class="hl-string">"REJECTED", class="hl-string">"RJCT"} for code in normalized_statuses) else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Authorized", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if class="hl-string">"AUTHORIZED" in normalized_statuses else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Instruction Sent", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if any(code in {class="hl-string">"SENT", class="hl-string">"DISPATCHED"} for code in normalized_statuses) else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Confirmed", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if any(code in {class="hl-string">"ACCP", class="hl-string">"ACCEPTED", class="hl-string">"CONFIRMED"} for code in normalized_statuses) else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Settled", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if any(code in {class="hl-string">"SETTLED", class="hl-string">"BOOK"} for code in normalized_statuses) else None
        add_event(events, seen, transaction_id, class="hl-string">"Payment Reconciled", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id) if class="hl-string">"RECONCILED" in normalized_statuses or any(code in normalized_reconciliation for code in {class="hl-string">"TRUE", class="hl-string">"MATCHED", class="hl-string">"RECONCILED"}) else None

    if any(code in ERROR_STATUS_CODES for code in normalized_statuses) or normalized_text:
        add_event(events, seen, transaction_id, class="hl-string">"Payment Error Identified", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id)
    if any(code in RESOLVED_STATUS_CODES for code in normalized_statuses) or any(code in normalized_repairs for code in {class="hl-string">"TRUE", class="hl-string">"RESOLVED", class="hl-string">"REPAIRED", class="hl-string">"CORRECTED"}):
        add_event(events, seen, transaction_id, class="hl-string">"Payment Error Resolved", event_time, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id)

    due = parse_datetime(due_date)
    settled = parse_datetime(settlement_date)
    if due and settled and settled.date() > due.date():
        add_event(events, seen, transaction_id, class="hl-string">"Late Payment Identified", settled, user, amount, currency, payment_type, due_date, settlement_date, business_unit, message_id)


def main() -> None:
    events: List[Dict[str, str]] = []
    seen = set()
    for path in sorted(INPUT_DIRECTORY.rglob(class="hl-string">"*.xml")):
        parse_file(path, events, seen)
    events.sort(key=lambda row: (row[class="hl-string">"PaymentTransactionId"], row[class="hl-string">"EventTimestamp"], row[class="hl-string">"ActivityName"]))
    OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
    with OUTPUT_FILE.open(class="hl-string">"w", newline=class="hl-string">"", encoding=class="hl-string">"utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=OUTPUT_COLUMNS)
        writer.writeheader()
        writer.writerows(events)
    print(fclass="hl-string">"Wrote {len(events)} event rows to {OUTPUT_FILE}")


if __name__ == class="hl-string">"__main__":
    main()