Written from 12 named sources USMA AOG Army A Club — Python Donor Reconciliation Pipeline (Final) This is the production-ready, audit-defensible reconciliation pipeline for migrating the Army A Club legacy donor database into PAC io Fundraising by Paciolan. It synthesizes all four improvement cycles, incorporates every fix from the Final Critique Report, and is grounded in the operational realities of the AOG's fundraising environment. Design Philosophy Using pandas and NumPy is the industry-standard Pythonic approach for data cleaning, deduplication, and transformation of this kind. The pipeline is organized around five commitments: Idempotency — Re-running on the same input always produces the same output. Audit provenance — Every change is traceable to a source file, a timestamp, and a reason code. Business-rule separation — Point calculation logic lives entirely in config, never hardcoded. Donor-ID primacy — Donor_ID is the stable foreign-key anchor because Army A Club benefits are determined first by membership level, then by priority-point standing. Points are recalculated accurately but are never used as the primary join key. Defensive output — The export is validated and hashed before upload so the PAC io bulk import [3][6] receives clean, verifiable data. Architecture Configuration Block ⚠️ Critical: The PRIORITY_POINT_FORMULA below is the canonical formula specified in the original requirements. Verify it against signed-off Army A Club documentation before running in production. The membership hierarchy and eligible gift types are examples and must be confirmed with AOG leadership. # config.py — load from a secure store (e.g., AWS Secrets Manager) in production CONFIG = { # ----------------------------------------------------------------------- # PRIORITY POINT FORMULA (must be verified against AOG documentation) # Available variables: eligible_donation_total, years_supported_max, # games_attended_total # ----------------------------------------------------------------------- "PRIORITY_POINT_FORMULA": ( "floor(eligible_donation_total / 100)" " + years_supported_max" " + (games_attended_total * 0.5)" ), # Gift types that count toward priority points [12] "ELIGIBLE_GIFT_TYPES": ["Unrestricted", "Annual Fund", "Margin Of Excellence"], # Membership hierarchy — higher value = higher tier [7] # Used by survivorship logic during aggregation. "MEMBERSHIP_HIERARCHY": { "Superintendent's Circle": 10, "Commandant's Circle": 9, "Army A Club": 5, "Unknown": 0, }, # Flag records whose recalculated points differ from legacy by more than this "POINT_DELTA_THRESHOLD": 2.0, # Basic email format check [8] "EMAIL_REGEX": r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", # Internal column → PAC io import header mapping [6] # Key order determines output column order. "PACIO_OUTPUT_SCHEMA": { "Final_ID": "CustomerId", "Full_Name_Clean": "Name", "Email_Clean": "EmailAddress", "Priority_Points_Calc": "PriorityPoints", "Eligible_Donation_Total":"LifetimeGivingAmount", "Entry_Date_Str": "AccountSetupDate", "Years_Supported_Max": "YearsOfService", "Membership_Level_Clean": "MembershipLevel", "Record_Hash": "ImportRecordHash", }, } Full Pipeline Code """ reconcile_pipeline.py ───────────────────────────────────────────────────────────────────────────── USMA AOG Army A Club — Legacy Donor Reconciliation Pipeline One-time migration: clean → aggregate → deduplicate → recalculate → export Outputs paciolan_import_<ts>.csv — PAC io bulk import file [3][6] audit_trail_<ts>.csv — Full provenance log [4] REVIEW_QUEUE_<ts>.csv — Records exceeding point-delta threshold Dependencies: pandas, numpy (pip install pandas numpy) """ from future import annotations import hashlib import logging import re from datetime import datetime from pathlib import Path import numpy as np import pandas as pd Configuration (paste CONFIG dict here, or import from config.py) CONFIG = { "PRIORITY_POINT_FORMULA": ( "floor(eligible_donation_total / 100)" " + years_supported_max" " + (games_attended_total * 0.5)" ), "ELIGIBLE_GIFT_TYPES": ["Unrestricted", "Annual Fund", "Margin Of Excellence"], "MEMBERSHIP_HIERARCHY": { "Superintendent's Circle": 10, "Commandant's Circle": 9, "Army A Club": 5, "Unknown": 0, }, "POINT_DELTA_THRESHOLD": 2.0, "EMAIL_REGEX": r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", "PACIO_OUTPUT_SCHEMA": { "Final_ID": "CustomerId", "Full_Name_Clean": "Name", "Email_Clean": "EmailAddress", "Priority_Points_Calc": "PriorityPoints", "Eligible_Donation_Total": "LifetimeGivingAmount", "Entry_Date_Str": "AccountSetupDate", "Years_Supported_Max": "YearsOfService", "Membership_Level_Clean": "MembershipLevel", "Record_Hash": "ImportRecordHash", }, } Logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) =========================================================================== Helper utilities =========================================================================== def get_file_hash(filepath: str) -> str: """SHA-256 hash of an on-disk file for source-level provenance.""" h = hashlib.sha256() try: with open(filepath, "rb") as fh: for chunk in iter(lambda: fh.read(4096), b""): h.update(chunk) return h.hexdigest() except FileNotFoundError: return "FILE_NOT_FOUND" def generate_row_hash(row: pd.Series, data_columns: list[str]) -> str: """ SHA-256 hash of the named data columns for idempotency verification. The hash column itself must NOT appear in data_columns. """ combined = "".join( str(row[c]) for c in data_columns if str(row[c]) not in ("nan", "NaT", "") ) return hashlib.sha256(combined.encode("utf-8")).hexdigest() def select_highest_membership(series: pd.Series) -> str float: """ Aggregation helper: returns the highest-ranked membership tier found across all transactions for one donor. [7] """ valid = series.dropna().unique() if len(valid) == 0: return np.nan return max(valid, key=lambda x: CONFIG["MEMBERSHIP_HIERARCHY"].get(x, 0)) def log_audit( audit_list: list[dict], row_id: str, field: str, old_val, new_val, reason: str, severity: str = "Info", ) -> None: audit_list.append( { "Timestamp": datetime.utcnow().isoformat(), "Record_ID": str(row_id), "Severity": severity, "Field": field, "Original_Value": str(old_val), "New_Value": str(new_val), "Reason": reason, } ) def _eval_formula(formula: str, kwargs: np.ndarray) -> np.ndarray: """ Evaluate a point-calculation formula string in a restricted namespace. Only numpy's floor and the explicitly supplied keyword-argument arrays are available — no builtins, no arbitrary attribute access. [security fix] """ safe_ns = {"__builtins__": {}, "floor": np.floor, kwargs} try: return eval(compile(formula, "<formula>", "eval"), safe_ns) # noqa: S307 except Exception as exc: raise ValueError( f"Formula evaluation failed.\n" f" Formula : {formula}\n" f" Variables available: {list(kwargs.keys())}\n" f" Error : {exc}" ) from exc =========================================================================== Phase 1 — Provenance & Ingestion =========================================================================== def load_with_provenance( legacy_path: str, attendance_path: str, audit_trail: list[dict], ) -> tuple[pd.DataFrame, pd.DataFrame]: """Load both source files and record their SHA-256 hashes in the audit trail.""" logger.info("Phase 1 · Loading data and establishing source provenance …") for path in (legacy_path, attendance_path): file_hash = get_file_hash(path) log_audit( audit_trail, "SYSTEM", "Source_File", path, file_hash, "Input file SHA-256 provenance hash established", "System", ) # Read as str to preserve leading zeros and avoid scientific notation [5][15] df_legacy = pd.read_csv(legacy_path, dtype=str) df_attendance = pd.read_csv(attendance_path, dtype=str) logger.info( f" Loaded {len(df_legacy)} legacy transaction rows " f"and {len(df_attendance)} attendance rows." ) return df_legacy, df_attendance =========================================================================== Phase 2 — Row-Level Cleaning & Normalization =========================================================================== def clean_normalize_rows( df: pd.DataFrame, audit_trail: list[dict], ) -> pd.DataFrame: """ Standardize every field at the individual transaction row level. [5][8][15] No aggregation occurs here — each row keeps its identity. """ logger.info("Phase 2 · Row-level cleaning and normalization …") df = df.copy() # ── 1. ID standardization ──────────────────────────────────────────── for col in ("Paciolan_ID", "Legacy_Donor_ID"): if col in df.columns: df[col] = df[col].replace({"nan": np.nan, "": np.nan}) else: df[col] = np.nan # Primary ID for aggregation: prefer existing Paciolan ID [6] df["Primary_ID"] = df["Paciolan_ID"].fillna(df["Legacy_Donor_ID"]) # ── 2. Email cleaning & format validation ──────────────────────────── df["Email_Clean"] = ( df["Email"].astype(str).str.lower().str.strip() .replace({"nan": np.nan, "": np.nan}) ) email_re = re.compile(CONFIG["EMAIL_REGEX"]) invalid_email_mask = df["Email_Clean"].notna() & ~df["Email_Clean"].apply( lambda x: bool(email_re.match(x)) ) for idx in df[invalid_email_mask].index: log_audit( audit_trail, df.loc[idx, "Primary_ID"], "Email", df.loc[idx, "Email"], df.loc[idx, "Email_Clean"], "Invalid email format detected — record retained but flagged", "Warning", ) # ── 3. Name, membership level & phone normalization ────────────────── df["Full_Name_Clean"] = ( df["Full_Name"].astype(str) .str.strip() .str.replace(r"[^\w\s]", "", regex=True) .str.title() ) membership_col = "Membership_Level" if "Membership_Level" in df.columns else None if membership_col: df["Membership_Level_Clean"] = ( df[membership_col].astype(str).str.strip().str.title() .replace({"nan": np.nan, "Nan": np.nan}) ) else: df["Membership_Level_Clean"] = np.nan if "Phone" in df.columns: df["Phone_Clean"] = ( df["Phone"].astype(str) .str.replace(r"\D", "", regex=True) .replace({"": np.nan}) ) else: df["Phone_Clean"] = np.nan # ── 4. Date coercion → NaT on failure [5] ──────────────────────────── df["Entry_Date_DT"] = pd.to_datetime(df["Entry_Date"], errors="coerce") # ── 5. Numeric coercion & non-negative validation [4] ──────────────── amount_col = "Donation_Amount" if "Donation_Amount" in df.columns else "Donation_Total" for col in (amount_col, "Years_Supported", "Priority_Points_Legacy"): if col not in df.columns: df[col] = 0.0 df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0.0) bad = df[df[col] < 0].index for idx in bad: log_audit( audit_trail, df.loc[idx, "Primary_ID"], col, df.loc[idx, col], 0, "Negative value corrected to zero", "Warning", ) df.loc[bad, col] = 0.0 # Normalise column name used downstream df["Donation_Amount"] = df[amount_col] # ── 6. Gift-type eligibility flagging (row level) ──────────────────── gift_col = "Gift_Type" if "Gift_Type" in df.columns else None if gift_col: df["Gift_Type_Clean"] = ( df[gift_col].astype(str).str.strip().str.title() ) df["Row_Eligible_Amount"] = np.where( df["Gift_Type_Clean"].isin(CONFIG["ELIGIBLE_GIFT_TYPES"]), df["Donation_Amount"], 0.0, ) else: logger.warning( " 'Gift_Type' column missing — treating ALL donations as eligible. " "Verify this business rule with Army A Club leadership." ) df["Gift_Type_Clean"] = "Unknown" df["Row_Eligible_Amount"] = df["Donation_Amount"] return df =========================================================================== Phase 3 — Transactional Aggregation & Multi-Stage Deduplication =========================================================================== def aggregate_and_deduplicate( df: pd.DataFrame, audit_trail: list[dict], ) -> pd.DataFrame: """ Collapse multiple transaction rows per donor into a single golden record, then remove cross-ID duplicates via a two-pass cascade. [15] """ logger.info("Phase 3 · Aggregating transactions and resolving donor identity …") # ── 1. Aggregate by Primary_ID ──────────────────────────────────────── agg_rules: dict = { "Paciolan_ID": "first", "Full_Name_Clean": "first", "Email_Clean": "first", "Phone_Clean": "first", "Membership_Level_Clean": select_highest_membership, # hierarchy rule [7] "Entry_Date_DT": "max", # keep latest date [15] "Years_Supported": "max", # keep maximum years "Priority_Points_Legacy":"first", # donor-level field, not per-transaction "Row_Eligible_Amount": "sum", # sum eligible donations across all txns } df_agg = df.groupby("Primary_ID", as_index=False).agg(agg_rules) df_agg = df_agg.rename( columns={ "Row_Eligible_Amount": "Eligible_Donation_Total", "Years_Supported": "Years_Supported_Max", } ) logger.info( f" Aggregated {len(df)} transaction rows " f"→ {len(df_agg)} unique Primary_ID records." ) initial_count = len(df_agg) # Sort preference: records that already have a Paciolan ID survive first, # then prefer the most recent entry date. NaT → oldest possible date so # real dates always win. [15] df_agg["_has_pacio"] = df_agg["Paciolan_ID"].notna().astype(int) df_agg["_sort_date"] = df_agg["Entry_Date_DT"].fillna(pd.Timestamp("1900-01-01")) df_agg = df_agg.sort_values( by=["_has_pacio", "_sort_date"], ascending=[False, True] ) # ── Pass 1: Email + Name deduplication ─────────────────────────────── df_deduped = df_agg.drop_duplicates( subset=["Email_Clean", "Full_Name_Clean"], keep="last" ) pass1_removed = initial_count - len(df_deduped) logger.info(f" Dedupe Pass 1 (Email + Name) removed {pass1_removed} records.") # ── Pass 2: Null-email fallback — Name + Phone + Membership Level ──── # Requiring all three columns significantly reduces false-positive merges # within households. if df_deduped["Phone_Clean"].notna().any(): phone_mask = df_deduped["Phone_Clean"].notna() df_has_phone = df_deduped[phone_mask].copy() df_no_phone = df_deduped[~phone_mask].copy() fallback_cols = ["Full_Name_Clean", "Phone_Clean", "Membership_Level_Clean"] df_phone_deduped = df_has_phone.drop_duplicates( subset=fallback_cols, keep="last" ) pass2_removed = len(df_has_phone) - len(df_phone_deduped) logger.info( f" Dedupe Pass 2 (Name + Phone + Membership) removed {pass2_removed} records." ) df_deduped = pd.concat([df_phone_deduped, df_no_phone], ignore_index=True) else: logger.info(" Dedupe Pass 2 skipped — no phone data present.") df_deduped = df_deduped.drop(columns=["_has_pacio", "_sort_date"]) # Stable output ID: prefer Paciolan ID where it exists [6] df_deduped["Final_ID"] = df_deduped["Paciolan_ID"].fillna(df_deduped["Primary_ID"]) return df_deduped =========================================================================== Phase 4 — Enrichment, Calculation & Business-Rule Validation =========================================================================== def enrich_calculate_validate( df_donor: pd.DataFrame, df_attendance: pd.DataFrame, audit_trail: list[dict], ) -> pd.DataFrame: """ Join game-attendance data, apply the canonical priority-point formula, then validate membership eligibility and flag point deltas. [2][7] """ logger.info("Phase 4 · Enriching, calculating points, and validating rules …") df = df_donor.copy() # ── 1. Join attendance (one attendance row per game played) ────────── id_col_att = ( "Legacy_Donor_ID" if "Legacy_Donor_ID" in df_attendance.columns else df_attendance.columns[0] ) att_counts = ( df_attendance .groupby(id_col_att) .size() .reset_index(name="games_attended_total") ) matched = df["Primary_ID"].isin(att_counts[id_col_att]).sum() logger.info( f" Attendance join: matched {matched} / {len(df)} donor records." ) df = df.merge( att_counts, left_on="Primary_ID", right_on=id_col_att, how="left", ) # Drop the merge artifact column (right-side key) if it was added if id_col_att in df.columns and id_col_att != "Primary_ID": df = df.drop(columns=[id_col_att], errors="ignore") df["games_attended_total"] = ( pd.to_numeric(df["games_attended_total"], errors="coerce").fillna(0.0) ) # ── 2. Priority-point calculation — restricted eval for security ───── eligible_donation_total = df["Eligible_Donation_Total"].to_numpy(dtype=float) years_supported_max = df["Years_Supported_Max"].to_numpy(dtype=float) games_attended_total = df["games_attended_total"].to_numpy(dtype=float) raw_points = _eval_formula( CONFIG["PRIORITY_POINT_FORMULA"], eligible_donation_total=eligible_donation_total, years_supported_max=years_supported_max, games_attended_total=games_attended_total, ) df["Priority_Points_Calc"] = np.round(raw_points, 2) logger.info(" Priority points recalculated using configured formula.") # ── 3. Membership-level validation [7] ─────────────────────────────── # Flag any record that has points but lacks a recognised membership tier. valid_levels = {k for k, v in CONFIG["MEMBERSHIP_HIERARCHY"].items() if v > 0} invalid_membership = ( (df["Priority_Points_Calc"] > 0) & ~df["Membership_Level_Clean"].isin(valid_levels) ) for idx in df[invalid_membership].index: log_audit( audit_trail, df.loc[idx, "Final_ID"], "Membership_Level", df.loc[idx, "Membership_Level_Clean"], "MISSING/INVALID", ( f"Points recalculated to {df.loc[idx, 'Priority_Points_Calc']} " "but no valid membership level found — benefits may not assign correctly." ), "Critical", ) # ── 4. Point-delta flag ────────────────────────────────────────────── df["Point_Delta"] = ( df["Priority_Points_Calc"] pd.to_numeric(df["Priority_Points_Legacy"], errors="coerce").fillna(0) ).abs() threshold = CONFIG["POINT_DELTA_THRESHOLD"] above_threshold = df["Point_Delta"] > threshold for idx in df[above_threshold].index: log_audit( audit_trail, df.loc[idx, "Final_ID"], "Priority_Points", df.loc[idx, "Priority_Points_Legacy"], df.loc[idx, "Priority_Points_Calc"], f"Delta {df.loc[idx, 'Point_Delta']:.2f} exceeded threshold {threshold}.", "Warning", ) df["Needs_Review"] = above_threshold flagged = above_threshold.sum() if flagged: logger.warning(f" {flagged} records flagged for manual review (point delta).") return df =========================================================================== Phase 5 — Finalization & Egress =========================================================================== def finalize_output(df: pd.DataFrame) -> pd.DataFrame: """ Map internal columns to PAC io import headers and stamp each row with a deterministic SHA-256 hash for idempotency verification. [6] """ logger.info("Phase 5 · Finalizing output schema and generating row hashes …") df = df.copy() # Format date; NaT → empty string for clean CSV output df["Entry_Date_Str"] = df["Entry_Date_DT"].apply( lambda x: x.strftime("%Y-%m-%d") if pd.notna(x) else "" ) schema_map = CONFIG["PACIO_OUTPUT_SCHEMA"] # Select only data columns that exist in the frame (exclude Record_Hash — # it hasn't been created yet; including it now causes a KeyError). data_cols = [ c for c in schema_map if c != "Record_Hash" and c in df.columns ] final_df = df[data_cols].copy() # Generate hash AFTER selecting data columns, BEFORE renaming. [critical fix] final_df["Record_Hash"] = final_df.apply( lambda row: generate_row_hash(row, data_cols), axis=1 ) # Rename to PAC io-facing headers final_df = final_df.rename(columns=schema_map) return final_df =========================================================================== Main orchestrator =========================================================================== def run_reconciliation_pipeline( legacy_file: str, attendance_file: str, output_dir: str, dry_run: bool = False, ) -> None: """ End-to-end pipeline orchestrator. Parameters legacy_file : Path to the legacy donor transaction CSV. attendance_file : Path to the attendance CSV (one row per game attended). output_dir : Directory where all output files are written. dry_run : If True, generate the audit log only — no import CSV. """ audit_trail: list[dict] = [] out_path = Path(output_dir) out_path.mkdir(parents=True, exist_ok=True) ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S") logger.info(f"{'='*60}") logger.info(f" Pipeline run started: {ts} dry_run={dry_run}") logger.info(f"{'='*60}") try: # 1 ─ Load df_raw, df_att = load_with_provenance(legacy_file, attendance_file, audit_trail) # 2 ─ Clean df_rows = clean_normalize_rows(df_raw, audit_trail) # 3 ─ Aggregate + deduplicate df_donors = aggregate_and_deduplicate(df_rows, audit_trail) # 4 ─ Enrich + calculate + validate df_calc = enrich_calculate_validate(df_donors, df_att, audit_trail) # 5 ─ Finalize schema + hash df_import = finalize_output(df_calc) # ── Write audit trail (Critical events sort to top) ─────────────── audit_df = pd.DataFrame(audit_trail) severity_order = {"Critical": 0, "Warning": 1, "Info": 2, "System": 3} audit_df = audit_df.sort_values( by="Severity", key=lambda s: s.map(severity_order).fillna(9) ) audit_file = out_path / f"audit_trail_{ts}.csv" audit_df.to_csv(audit_file, index=False) logger.info(f" Audit trail → {audit_file} ({len(audit_df)} entries)") # ── Write review queue ──────────────────────────────────────────── review_df = df_calc[df_calc["Needs_Review"]] if not review_df.empty: review_file = out_path / f"REVIEW_QUEUE_{ts}.csv" review_df.to_csv(review_file, index=False) logger.warning( f" {len(review_df)} records need manual review → {review_file}" ) # ── Write import file (skipped in dry-run mode) ─────────────────── if dry_run: logger.info(" DRY RUN: import file not written.") else: # Strip any trailing ".0" from ID columns caused by NaN coercion [5] pacio_id_col = CONFIG["PACIO_OUTPUT_SCHEMA"]["Final_ID"] if pacio_id_col in df_import.columns: df_import[pacio_id_col] = ( df_import[pacio_id_col] .astype(str) .str.replace(r"\.0$", "", regex=True) ) import_file = out_path / f"paciolan_import_{ts}.csv" df_import.to_csv(import_file, index=False) logger.info(f" PAC io import file → {import_file}") logger.info(" Pipeline completed successfully.") except Exception: logger.critical(" Pipeline failed — see traceback below.", exc_info=True) raise =========================================================================== Entry point with self-contained test data =========================================================================== if name == "__main__": # ── Generate illustrative dummy data ────────────────────────────────── logger.info("Generating dummy test data …") with open("dummy_legacy_transactions.csv", "w") as f: f.write( "Legacy_Donor_ID,Paciolan_ID,Full_Name,Email,Phone," "Membership_Level,Priority_Points_Legacy," "Donation_Amount,Gift_Type,Entry_Date,Years_Supported\n" ) # Donor 1 — two transactions; mixed membership; should survive as # "Superintendent's Circle" (higher tier). Eligible total = $5,000. f.write("L123,,John Smith,J.SMITH@email.com,5550100," "Army A Club,100,5000,Unrestricted,2020-01-15,5\n") f.write("L123,,John Smith,J.SMITH@email.com,5550100," "Superintendent's Circle,100,2500,Sport-Specific,2021-01-15,6\n") # Donor 2 — existing Paciolan ID; straightforward migration. f.write("L124,P999,Jane Doe,jane.doe@email.com,5550101," "Commandant's Circle,250,12000,Annual Fund,2018-05-20,8\n") # Donors 3 & 4 — null email, same name+phone but DIFFERENT membership: # must NOT be merged by Pass 2 (different tier = different person risk). f.write("L126,,Robert Jones,,5559999," "Army A Club,50,500,Unrestricted,2022-01-01,2\n") f.write("L127,,Robert Jones,,5559999," "Superintendent's Circle,50,500,Unrestricted,2022-06-01,2\n") # Donor 5 — missing membership; $1,000 eligible → triggers Critical audit. f.write("L128,,Missing Member,mm@email.com,5551111," ",0,1000,Unrestricted,2023-01-01,1\n") with open("dummy_attendance.csv", "w") as f: f.write("Legacy_Donor_ID,Game_Date\n") f.write("L123,2023-09-01\n") f.write("L123,2023-09-15\n") f.write("L126,2023-09-01\n") run_reconciliation_pipeline( legacy_file="dummy_legacy_transactions.csv", attendance_file="dummy_attendance.csv", output_dir="./output_final", dry_run=False, # set True to test without writing the import file ) Inputs & Outputs Required input files File Minimum columns Legacy donor transactions CSV Legacy_Donor_ID, Full_Name, Email, Donation_Amount, Gift_Type, Entry_Date, Years_Supported, Priority_Points_Legacy Attendance CSV Legacy_Donor_ID, one row per game att