65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
|
|
def _stable_default(value: Any) -> str:
|
|
if isinstance(value, datetime | date):
|
|
return value.isoformat()
|
|
if isinstance(value, Decimal):
|
|
return str(value)
|
|
return str(value)
|
|
|
|
|
|
def _stable_json(value: Any) -> str:
|
|
return json.dumps(
|
|
value, default=_stable_default, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
)
|
|
|
|
|
|
def compute_raw_observation_hash(
|
|
*,
|
|
source_type: str,
|
|
source_reference: str | None = None,
|
|
raw_text: str | None = None,
|
|
raw_payload: dict[str, Any] | None = None,
|
|
) -> str:
|
|
payload = {
|
|
"source_reference": source_reference,
|
|
"source_type": source_type,
|
|
"raw_payload": raw_payload or {},
|
|
"raw_text": raw_text or "",
|
|
}
|
|
return hashlib.sha256(_stable_json(payload).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def normalize_counterparty(value: str | None) -> str:
|
|
if not value:
|
|
return ""
|
|
compact = re.sub(r"[^a-z0-9]+", " ", value.lower())
|
|
return re.sub(r"\s+", " ", compact).strip()
|
|
|
|
|
|
def compute_normalized_transaction_fingerprint(
|
|
*,
|
|
amount: Decimal | str | int | float,
|
|
currency: str,
|
|
event_date: date | str,
|
|
merchant: str | None = None,
|
|
counterparty: str | None = None,
|
|
external_id: str | None = None,
|
|
) -> str:
|
|
normalized = {
|
|
"amount": str(Decimal(str(amount)).quantize(Decimal("0.01"))),
|
|
"currency": currency.upper(),
|
|
"event_date": event_date.isoformat() if isinstance(event_date, date) else event_date,
|
|
"party": normalize_counterparty(merchant or counterparty),
|
|
"external_id": external_id or "",
|
|
}
|
|
return hashlib.sha256(_stable_json(normalized).encode("utf-8")).hexdigest()
|