# Phase Recon-1 — Python Matching Engine (CloudI Service)

**Status:** ⬜ Pending
**Duration:** Weeks 3–4
**Depends on:** Phase Recon-0
**Goal:** `recon_engine_py` CloudI service handles all reconciliation match types. Elixir calls it via `MwRecon.EngineClient`. No UI yet — tested via IEx and unit tests.

---

## Service Structure

```
services/recon_engine_py/
├── main.py               # CloudI service entry — subscribe + dispatch
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
├── normaliser.py         # Bank column-name → canonical name mapping
├── matchers/
│   ├── __init__.py
│   ├── base.py           # Shared merge/dedup utilities
│   ├── card.py           # Bank Card vs MomentsPay (4-key → 3-key fallback)
│   ├── upi.py            # Bank UPI vs MomentsPay
│   ├── his_bank_card.py  # HIS vs Bank Card statement
│   ├── his_bank_upi.py   # HIS vs Bank UPI statement
│   ├── his_momentspay.py # HIS vs MomentsPay (3-source)
│   └── amex.py           # AMEX statement vs HIS
└── tests/
    ├── test_normaliser.py
    ├── test_card.py
    ├── test_upi.py
    ├── test_his_bank.py
    └── fixtures/         # Sample rows extracted from reference Python scripts
```

---

## CloudI Service: `main.py`

```python
"""
recon_engine_py — Reconciliation matching engine.

Handles all hospital × payment-type reconciliation logic.
Accepts file payloads as base64, returns matched/unmatched row lists.

CloudI path:  /+/recon/match/
              "+" matches any tenant_id

cloudi.conf entry:
  [{prefix,        "/+/recon/match/"},
   {file_path,     "/usr/bin/python3"},
   {args,          "/app/services/recon_engine_py/main.py"},
   {count_process, 2},
   {max_r,         5},
   {max_t,         60},
   {env,           [{"PYTHONPATH", "/usr/local/lib/cloudi-2.0.7/api/python"}]}
  ]

Request payload (JSON):
  {
    "recon_type": "bank_card_vs_momentspay",
    "recon_date": "2025-06-09",          # ISO date string — used for HIS date filter
    "config": { ... },                    # ReconConfig fields (column_mappings, etc.)
    "files": {
      "bank_card":   "<base64 xlsx>",
      "momentspay":  "<base64 csv>",
      "his":         "<base64 xlsx>"      # Only for HIS-type recons
    }
  }

Response (JSON):
  {
    "status": "ok",
    "recon_type": "bank_card_vs_momentspay",
    "matched":   [ { ...row fields... }, ... ],
    "unmatched": [ { ...row fields... }, ... ],
    "summary": {
      "total": 120,
      "matched": 110,
      "unmatched": 10,
      "matched_amount": "135420.00",
      "unmatched_amount": "1230.00",
      "by_location": {
        "BBW": {"total": 20, "matched": 18, "unmatched": 2, ...},
        ...
      }
    }
  }
"""
import sys, json, logging, base64, io
sys.path.append('/usr/local/lib/cloudi-2.0.7/api/python/')
from cloudi import API, terminate_exception

from matchers.card          import CardMatcher
from matchers.upi           import UpiMatcher
from matchers.his_bank_card import HisBankCardMatcher
from matchers.his_bank_upi  import HisBankUpiMatcher
from matchers.his_momentspay import HisMomentspayMatcher
from matchers.amex          import AmexMatcher

MATCHER_MAP = {
    "bank_card_vs_momentspay":  CardMatcher,
    "bank_upi_vs_momentspay":   UpiMatcher,
    "his_bank_card":            HisBankCardMatcher,
    "his_bank_upi":             HisBankUpiMatcher,
    "his_momentspay_card":      HisMomentspayMatcher,
    "his_momentspay_upi":       HisMomentspayMatcher,
    "amex":                     AmexMatcher,
}

logger = logging.getLogger('recon_engine_py')

class ReconEngineService:
    def __init__(self):
        self.__api = API(0)

    def run(self):
        try:
            self.__api.subscribe("+/recon/match/", self.__handle)
            logger.info("recon_engine_py started")
            self.__api.poll()
        except terminate_exception:
            logger.info("recon_engine_py terminated cleanly")

    def __handle(self, request_type, name, pattern, request_info, request,
                 timeout, priority, trans_id, pid):
        try:
            payload = json.loads(request)
            recon_type = payload["recon_type"]
            matcher_cls = MATCHER_MAP.get(recon_type)
            if not matcher_cls:
                return json.dumps({"status": "error", "reason": f"Unknown recon_type: {recon_type}"})
            matcher = matcher_cls(payload)
            result = matcher.run()
            return json.dumps({"status": "ok", **result})
        except Exception as e:
            logger.error(f"Match failed: {e}", exc_info=True)
            return json.dumps({"status": "error", "reason": str(e)})

if __name__ == '__main__':
    assert API.thread_count() == 1
    ReconEngineService().run()
```

---

## Column Normaliser: `normaliser.py`

Handles the fact that each bank uses different column names for the same concept.
Config is loaded from the `ReconConfig.column_mappings` JSON field.

```python
"""
normaliser.py — Maps bank-specific column names to canonical names.

Default mappings (can be overridden per hospital via ReconConfig.column_mappings):

  canonical        | bank aliases
  -----------------|------------------------------------------------------------
  CARDNBR          | CARD NUMBER, PAYER VPA, Payer VPA
  TERMINAL_NO      | TERMINAL NUMBER, TerminalID, EXTERNAL TID
  AUTH_AMOUNT      | Transaction Amount, TransactionAmount, Amount, DOMESTIC AMT
  APPROVAL_CODE    | APPROV CODE, Approval No, Approval Code
  RRN_NO           | Txn ref no.(RRN), RRN
  TRANS_DATE       | TRANS DATE, Transaction Date, Value Date
"""
import pandas as pd

DEFAULT_COLUMN_MAPPING = {
    'CARDNBR':       ['CARD NUMBER', 'PAYER VPA', 'Payer VPA', 'Card No/Cheque'],
    'TERMINAL_NO':   ['TERMINAL NUMBER', 'TerminalID', 'EXTERNAL TID', 'TERMINAL_NO'],
    'AUTH_AMOUNT':   ['Transaction Amount', 'TransactionAmount', 'Amount',
                      'DOMESTIC AMT', 'AUTH_AMOUNT'],
    'APPROVAL_CODE': ['APPROV CODE', 'Approval No', 'Approval Code', 'APPROV_CODE'],
    'RRN_NO':        ['Txn ref no.(RRN)', 'RRN', 'RRN_NO'],
    'TRANS_DATE':    ['TRANS DATE', 'Transaction Date', 'Value Date', 'Date'],
}

def normalise(df: pd.DataFrame, extra_mappings: dict = None) -> pd.DataFrame:
    """
    Apply column normalisation to a bank DataFrame.
    Returns a copy with canonical column names added.
    Does not remove original columns.
    """
    mapping = {**DEFAULT_COLUMN_MAPPING, **(extra_mappings or {})}
    df = df.copy()
    for canonical, aliases in mapping.items():
        if canonical not in df.columns:
            actual = next((col for col in aliases if col in df.columns), None)
            if actual:
                df[canonical] = df[actual]
    return df
```

---

## Card Matcher: `matchers/card.py`

Ports logic from `SAHYADRIBANKCARDVSMOMENTPAYnewconsidering2combinationalso.py`:

```python
"""
card.py — Bank Card statement vs MomentsPay.

Match keys (4-key primary):
  last 4 digits of card  +  terminal_id  +  amount  +  approval_code

Fallback (3-key):
  last 4 digits of card  +  terminal_id  +  amount

Handles DOMESTIC vs INTERNATIONAL amount selection.
"""
import pandas as pd
import base64, io
from normaliser import normalise

class CardMatcher:
    def __init__(self, payload):
        self.config      = payload.get("config", {})
        self.recon_date  = payload.get("recon_date")
        bank_b64         = payload["files"]["bank_card"]
        mp_b64           = payload["files"]["momentspay"]
        self.bank_df     = pd.read_excel(io.BytesIO(base64.b64decode(bank_b64)))
        self.mp_df       = pd.read_csv(io.StringIO(base64.b64decode(mp_b64).decode('utf-8')))

    def run(self) -> dict:
        bank = self._prep_bank()
        mp   = self._prep_momentspay()

        # Primary: 4-key match
        keys_4 = ['Last4_CARDNBR', 'TERMINAL_NO', 'AUTH_AMOUNT', 'APPROVAL_CODE']
        mp_keys_4 = ['Last4_card_num', 'terminal_id', 'total_amount', 'approval_code']

        merged = pd.merge(
            bank, mp[['processing_id', 'transaction_id', 'email'] + mp_keys_4],
            left_on=keys_4, right_on=mp_keys_4, how='left'
        )
        merged['MomentsPay Matched'] = merged['processing_id'].notna().map({True: 'YES', False: 'NO'})

        # Fallback: 3-key match for unmatched rows
        unmatched_mask = merged['MomentsPay Matched'] == 'NO'
        if unmatched_mask.any():
            keys_3    = ['Last4_CARDNBR', 'TERMINAL_NO', 'AUTH_AMOUNT']
            mp_keys_3 = ['Last4_card_num', 'terminal_id', 'total_amount']
            unmatched = merged[unmatched_mask].drop(
                columns=['MomentsPay Matched', 'processing_id', 'transaction_id', 'email'],
                errors='ignore'
            )
            fallback = pd.merge(
                unmatched,
                mp[['processing_id', 'transaction_id', 'email'] + mp_keys_3],
                left_on=keys_3, right_on=mp_keys_3, how='left'
            )
            fallback['MomentsPay Matched'] = fallback['processing_id'].notna().map({True: 'YES', False: 'NO'})
            merged = pd.concat([merged[~unmatched_mask], fallback], ignore_index=True)

        merged = merged.dropna(subset=['CARD NUMBER'])
        matched   = merged[merged['MomentsPay Matched'] == 'YES']
        unmatched = merged[merged['MomentsPay Matched'] == 'NO']

        return {
            "recon_type": "bank_card_vs_momentspay",
            "matched":    matched.to_dict('records'),
            "unmatched":  unmatched.to_dict('records'),
            "summary":    self._summarise(matched, unmatched)
        }

    def _prep_bank(self) -> pd.DataFrame:
        df = normalise(self.bank_df, self.config.get("column_mappings"))
        df['CARD NUMBER']   = df['CARD NUMBER'].astype(str).str.replace(r'\.0$', '', regex=True)
        df['TERMINAL_NO']   = df['TERMINAL_NO'].astype(str).str.replace(r'\.0$', '', regex=True).str.replace(r"^'", '', regex=True)
        df['APPROVAL_CODE'] = df['APPROVAL_CODE'].astype(str).str.replace(r'\.0$', '', regex=True).str.replace(r"'", '', regex=True)
        # DOMESTIC vs INTERNATIONAL amount
        if 'INTNL AMT' in df.columns and 'DOMESTIC AMT' in df.columns:
            df['INTNL AMT']   = df['INTNL AMT'].astype(str).str.replace(r'\.0$', '', regex=True)
            df['DOMESTIC AMT'] = df['DOMESTIC AMT'].astype(str).str.replace(r'\.0$', '', regex=True)
            df['AUTH_AMOUNT'] = df.apply(
                lambda r: r['DOMESTIC AMT'] if r['INTNL AMT'] == '0' else r['INTNL AMT'], axis=1
            )
        else:
            df['AUTH_AMOUNT'] = df['AUTH_AMOUNT'].astype(str).str.replace(r'\.0$', '', regex=True)
        # Pad approval code to 6
        df['APPROVAL_CODE'] = df['APPROVAL_CODE'].apply(lambda x: x.zfill(6) if len(x) != 6 else x)
        # Last 4 of card
        df['Last4_CARDNBR'] = df['CARD NUMBER'].str[-4:]
        # Exclude header rows
        exclude = ['Bank PAN No', 'Merchant PAN No.', 'Curr Ac GSTN', 'Acc No2 GSTN']
        df = df[~df.iloc[:, 0].astype(str).str.startswith(tuple(exclude))]
        return df

    def _prep_momentspay(self) -> pd.DataFrame:
        df = self.mp_df.copy()
        df['card_num']      = df['card_num'].astype(str).str.replace(r'\.0$', '', regex=True)
        df['terminal_id']   = df['terminal_id'].astype(str).str.replace(r'\.0$', '', regex=True)
        df['total_amount']  = df['total_amount'].astype(str).str.replace(r'\.0$', '', regex=True)
        df['approval_code'] = df['approval_code'].astype(str).str.replace(r'\.0$', '', regex=True).apply(lambda x: x.zfill(6))
        df['Last4_card_num'] = df['card_num'].str[-4:]
        return df

    def _summarise(self, matched, unmatched) -> dict:
        total_matched   = len(matched)
        total_unmatched = len(unmatched)
        total           = total_matched + total_unmatched

        def safe_sum(df, col):
            try:
                return float(df[col].astype(float).sum())
            except Exception:
                return 0.0

        return {
            "total":             total,
            "matched":           total_matched,
            "unmatched":         total_unmatched,
            "matched_amount":    str(safe_sum(matched, 'AUTH_AMOUNT')),
            "unmatched_amount":  str(safe_sum(unmatched, 'AUTH_AMOUNT')),
        }
```

---

## UPI Matcher: `matchers/upi.py`

Ports logic from `SAHYADRIBANKUPIVSMOMENTPAY.py`:

```python
"""
upi.py — Bank UPI statement vs MomentsPay.

Match keys: CARDNBR (VPA), TERMINAL_NO, AUTH_AMOUNT, RRN_NO
RRN zero-padded to 12 digits.
"""
import pandas as pd, base64, io
from normaliser import normalise

class UpiMatcher:
    def __init__(self, payload):
        self.config = payload.get("config", {})
        bank_b64 = payload["files"]["bank_upi"]
        mp_b64   = payload["files"]["momentspay"]
        self.bank_df = pd.read_excel(io.BytesIO(base64.b64decode(bank_b64)))
        self.mp_df   = pd.read_csv(io.StringIO(base64.b64decode(mp_b64).decode('utf-8')))

    def run(self):
        bank = normalise(self.bank_df, self.config.get("column_mappings"))
        mp   = self.mp_df.copy()

        bank['CARDNBR']   = bank['CARDNBR'].astype(str).str.replace(r'\.0$', '', regex=True)
        bank['TERMINAL_NO'] = bank['TERMINAL_NO'].astype(str).str.replace(r'\.0$', '', regex=True).str.replace(r"^'", '', regex=True)
        bank['AUTH_AMOUNT'] = bank['AUTH_AMOUNT'].astype(str).str.replace(r'\.0$', '', regex=True)
        bank['RRN_NO']    = bank['RRN_NO'].astype(str).str.replace(r'\.0$', '', regex=True).str.replace(r"'", '', regex=True)

        mp['card_num']    = mp['card_num'].astype(str).str.replace(r'\.0$', '', regex=True)
        mp['terminal_id'] = mp['terminal_id'].astype(str).str.replace(r'\.0$', '', regex=True)
        mp['total_amount'] = mp['total_amount'].astype(str).str.replace(r'\.0$', '', regex=True)
        mp['rrn_no']      = mp['rrn_no'].astype(str).str.replace(r'\.0$', '', regex=True).apply(lambda x: x.zfill(12))

        merged = pd.merge(
            bank, mp[['processing_id', 'transaction_id', 'email', 'card_num', 'terminal_id', 'total_amount', 'rrn_no']],
            left_on =['CARDNBR', 'TERMINAL_NO', 'AUTH_AMOUNT', 'RRN_NO'],
            right_on=['card_num', 'terminal_id', 'total_amount', 'rrn_no'],
            how='left'
        )
        merged['MomentsPay Matched'] = merged['processing_id'].notna().map({True: 'YES', False: 'NO'})
        merged = merged.dropna(subset=['CARDNBR'])

        matched   = merged[merged['MomentsPay Matched'] == 'YES']
        unmatched = merged[merged['MomentsPay Matched'] == 'NO']
        return {
            "recon_type": "bank_upi_vs_momentspay",
            "matched":    matched.to_dict('records'),
            "unmatched":  unmatched.to_dict('records'),
            "summary":    {"total": len(merged), "matched": len(matched), "unmatched": len(unmatched)}
        }
```

---

## HIS-Bank Card Matcher: `matchers/his_bank_card.py`

Ports logic from `MSBC.py` (Sahyadri HIS Bank Card):

```python
"""
his_bank_card.py — HIS export vs Bank Card statement.

Match keys: last4 card + amount + approval_code
Date filter applied to HIS data before matching.
Location mapping applied from config.
"""
# (see full implementation in phase detail doc — abbreviated here)
class HisBankCardMatcher:
    def __init__(self, payload):
        self.config     = payload.get("config", {})
        self.recon_date = payload.get("recon_date")
        his_b64  = payload["files"]["his"]
        bank_b64 = payload["files"]["bank_card"]
        import pandas as pd, base64, io
        self.his_df  = pd.read_excel(io.BytesIO(base64.b64decode(his_b64)))
        self.bank_df = pd.read_excel(io.BytesIO(base64.b64decode(bank_b64)))

    def run(self):
        # Filter HIS to recon_date + payment mode = Debit Card / Credit Card
        # Apply location mapping from config.location_map
        # Match on last4 + amount + approval_code
        # Return matched/unmatched + location-grouped summary
        ...
```

---

## Elixir Client: `apps/mw_recon/lib/mw_recon/engine_client.ex`

```elixir
defmodule MwRecon.EngineClient do
  @moduledoc """
  Calls the recon_engine_py CloudI service.
  Serialises file payloads and config; deserialises match results.
  """

  alias AdapterCloudi.Dispatcher

  @service_path "/recon/match/"
  @timeout_ms   120_000  # 2 minutes — large files may take time

  @spec run_match(map()) :: {:ok, map()} | {:error, String.t()}
  def run_match(payload) when is_map(payload) do
    case Dispatcher.call(
           "#{payload.tenant_id}#{@service_path}",
           Jason.encode!(payload),
           @timeout_ms
         ) do
      {:ok, response_json} ->
        case Jason.decode(response_json) do
          {:ok, %{"status" => "ok"} = result}    -> {:ok, result}
          {:ok, %{"status" => "error", "reason" => r}} -> {:error, r}
          _                                       -> {:error, "Invalid response from engine"}
        end
      {:error, reason} ->
        {:error, "CloudI call failed: #{inspect(reason)}"}
    end
  end

  @doc """
  Builds the payload map from session + uploaded file binaries.
  """
  @spec build_payload(map(), map(), String.t()) :: map()
  def build_payload(session, files_b64, tenant_id) do
    %{
      tenant_id:  tenant_id,
      recon_type: session.recon_type,
      recon_date: Date.to_iso8601(session.recon_date),
      config:     session_config(session),
      files:      files_b64
    }
  end

  defp session_config(session) do
    config = session.config || %{}
    %{
      column_mappings: decode_json(config[:column_mappings]),
      location_map:    decode_json(config[:location_map]),
      match_keys:      decode_json(config[:match_keys]),
      payment_modes:   decode_json(config[:payment_modes])
    }
  end

  defp decode_json(nil), do: %{}
  defp decode_json(s) when is_binary(s), do: Jason.decode!(s)
  defp decode_json(m) when is_map(m), do: m
end
```

---

## `config/cloudi.conf` — Add Entries

```erlang
%% recon_engine_py
{prefix,        "/+/recon/match/"},
{file_path,     "/usr/bin/python3"},
{args,          "/app/services/recon_engine_py/main.py"},
{count_process, 2},
{max_r,         5},
{max_t,         60},
{env,           [{"PYTHONPATH", "/usr/local/lib/cloudi-2.0.7/api/python"}]}
```

---

## Acceptance Criteria

- [ ] All 5 matcher types return `{status: "ok", matched: [...], unmatched: [...], summary: {...}}`
- [ ] Card 4-key → 3-key fallback correctly picks up rows that only match on 3 keys
- [ ] Column normalisation maps all bank column variants from all 7 hospital scripts
- [ ] Unit tests cover each matcher with reference sample data
- [ ] Service restarts within 3s when killed
- [ ] `MwRecon.EngineClient.run_match/1` from IEx returns correct results