Written from 4 named sources This briefing details a complete, production-ready ETL (Extract, Transform, Load) pipeline for the United States Military Academy (USMA) Association of Graduates (AOG). Priority points are the lifeblood of athletic donor relations, determining equitable access to seating, parking, and post-season tickets [4], calbears.com. For the Army A Club, which funds the Academy's "Margin of Excellence" westpointaog.org, maintaining the integrity of these points is critical for donor trust. This pipeline automates the reconciliation of manually entered priority points against calculated ground-truth values. Incorporating higher-ed data hygiene best practices [1][3][6], it utilizes a GET-before-PATCH pattern, environment-based configuration, and a strict SQLite rollback log to ensure donor data is safely and accurately synchronized with Paciolan Spectra. Test Data (sample_donors.csv) Save this file locally to test the pipeline's edge cases (clean records, unverified notes, math mismatches, and malformed JSON) without risking live data. donor_id,name,manual_priority_points,adjustment_notes,donation_history 10001,Nimitz Chester,150,,[{"amount": 10000, "date": "2024-05-01"}] 10002,MacArthur Douglas,210,"+50 service bonus","[{""amount"": 16000, ""date"": ""2024-01-15""}]" 10003,Patton George,500,Missing adjustment amount,"[{""amount"": 5000, ""date"": ""2023-11-11""}]" 10004,Eisenhower Dwight,375,,[{"amount": 10000, "date": "2024-02-01"}, {"amount": 15000, "date": "2023-02-01"}] 10005,Bradley Omar,999,,Malformed JSON string Python ETL Script (priority_points_etl.py) This script executes the dual-stage reconciliation. It uses argparse for CLI execution and python-dotenv to secure API tokens. """ priority_points_etl.py One-time cleanup ETL for Army A Club priority points. Reconciles manual entries against calculated ground-truth and pushes verified corrections to Paciolan Spectra via REST API. Requirements (requirements.txt): pandas==2.1.0 requests==2.31.0 python-dotenv==1.0.0 sqlite3 (standard library) """ import argparse import ast import json import os import re import sqlite3 import time from datetime import datetime import pandas as pd import requests from dotenv import load_dotenv --- CONFIGURATION --- REQUIRED_COLUMNS = ["donor_id", "name", "donation_history", "manual_priority_points"] TIMEOUT_SECONDS = 10 def load_config(): """Load API configuration from environment variables.""" load_dotenv() return { "api_base_url": os.getenv("SPECTRA_API_BASE_URL", "https://api.paciolan.com").rstrip('/'), "auth_token": os.getenv("SPECTRA_AUTH_TOKEN", "YOUR_BEARER_TOKEN_HERE"), # ASSUMPTION: Verify this exact endpoint path with Paciolan Spectra API documentation "endpoint_template": os.getenv("SPECTRA_ENDPOINT_TEMPLATE", "/api/v1/donors/{donor_id}/priority-points"), "db_path": os.getenv("ETL_RUN_LOG_DB", "etl_run_log.db") } def coerce_amount_to_float(amount): """Convert donation amount values into float dollars.""" if pd.isna(amount) or amount is None: return 0.0 text = str(amount).strip() cleaned = re.sub(r"[^0-9.\-]", "", text) try: return float(cleaned) if cleaned else 0.0 except ValueError: return 0.0 def normalize_donation_history(value): """Normalize donation_history string into a list of dicts.""" if pd.isna(value) or not value: return [] parsed = value if isinstance(value, str): try: parsed = json.loads(value.strip()) except Exception: try: parsed = ast.literal_eval(value.strip()) except Exception: return [] if isinstance(parsed, dict): parsed = [parsed] normalized = [] if isinstance(parsed, list): for item in parsed: if isinstance(item, dict) and "amount" in item and "date" in item: normalized.append({ "amount": coerce_amount_to_float(item["amount"]), "date": str(item["date"]).strip() }) return normalized def calculate_base_points(donation_history): """ Calculates points: 1 per $100 (floored per transaction) with tenure multiplier. ASSUMPTION: This specific formula must be explicitly approved by USMA AOG leadership, as the 2025 Army A Club Benefits Chart prioritizes membership level first. """ if not donation_history: return 0 transaction_points = sum(int(coerce_amount_to_float(g.get("amount")) // 100) for g in donation_history) # Calculate continuous years ending at most recent gift years = sorted({datetime.strptime(g["date"], "%Y-%m-%d").year for g in donation_history if g.get("date")}, reverse=True) continuous = 0 if years: expected = years[0] for yr in years: if yr == expected: continuous += 1 expected -= 1 else: break multiplier = 1.5 if continuous >= 10 else (1.25 if continuous >= 5 else 1.0) return int(transaction_points * multiplier) def parse_adjustment_delta(note): """Extract signed numeric adjustments from free-text notes (e.g., '+50 service bonus').""" if pd.isna(note) or not str(note).strip(): return 0 match = re.search(r"([+-]\s*\d+)", str(note)) return int(match.group(1).replace(" ", "")) if match else 0 def reconcile_record(row): """Classify a donor row as CLEAN, WARNING, or ERROR.""" calc = calculate_base_points(row["donation_history"]) adj = parse_adjustment_delta(row.get("adjustment_notes", "")) corrected_points = calc + adj if pd.isna(row.get("manual_priority_points")): return "ERROR", adj, calc, corrected_points, "Missing manual_priority_points" manual = int(row["manual_priority_points"]) delta = manual - corrected_points has_note = pd.notna(row.get("adjustment_notes")) and str(row["adjustment_notes"]).strip() != "" if delta == 0: return "CLEAN", adj, calc, corrected_points, "Matches calculated value" if has_note and adj == 0: return "WARNING", adj, calc, corrected_points, "Unverified adjustment note present" if has_note: return "ERROR", adj, calc, corrected_points, f"Math does not reconcile; off by {delta}" return "ERROR", adj, calc, corrected_points, f"Manual points off by {delta} with no valid adjustment" def api_request(method, url, headers, payload=None): """Wrapper for API calls with basic retry logic on 429/503.""" for attempt in range(2): try: if method.upper() == "GET": resp = requests.get(url, headers=headers, timeout=TIMEOUT_SECONDS) else: resp = requests.patch(url, json=payload, headers=headers, timeout=TIMEOUT_SECONDS) if resp.status_code in (429, 503) and attempt == 0: time.sleep(1) continue return resp except requests.RequestException: pass return None def process_api_updates(errors_df, config): """Perform GET-before-PATCH and log outcomes to SQLite for rollback reference.""" headers = {"Authorization": f"Bearer {config['auth_token']}", "Content-Type": "application/json"} conn = sqlite3.connect(config["db_path"]) cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS run_log ( donor_id TEXT, timestamp TEXT, spectra_old_value INTEGER, new_value INTEGER, api_status_code INTEGER, success BOOLEAN ) """) conn.commit() success_count, fail_count = 0, 0 for _, row in errors_df.iterrows(): donor_id = str(row["donor_id"]) target_points = row["corrected_points"] url = config["api_base_url"] + config["endpoint_template"].format(donor_id=donor_id) # 1. GET current state from Spectra get_resp = api_request("GET", url, headers) spectra_old_value = None if get_resp and get_resp.status_code == 200: # ASSUMPTION: verify exact JSON response structure (e.g., {"priorityPoints": 150}) spectra_old_value = get_resp.json().get("priorityPoints") # Skip if already correct in Spectra if spectra_old_value == target_points: continue # 2. PATCH new state patch_resp = api_request("PATCH", url, headers, payload={"priorityPoints": target_points}) status_code = patch_resp.status_code if patch_resp else 999 success = 200 <= status_code < 300 # 3. Log to SQLite cur.execute(""" INSERT INTO run_log (donor_id, timestamp, spectra_old_value, new_value, api_status_code, success) VALUES (?, ?, ?, ?, ?, ?) """, (donor_id, datetime.now().isoformat(), spectra_old_value, target_points, status_code, success)) conn.commit() if success: success_count += 1 else: fail_count += 1 conn.close() return success_count, fail_count def main(): parser = argparse.ArgumentParser(description="Army A Club Priority Points ETL") parser.add_argument("--source", required=True, help="Path to source CSV") parser.add_argument("--dry-run", action="store_true", help="Preview changes without API updates") args = parser.parse_args() config = load_config() df = pd.read_csv(args.source) # Ensure columns exist if "adjustment_notes" not in df.columns: df["adjustment_notes"] = "" df["donation_history"] = df["donation_history"].apply(normalize_donation_history) results = [] for _, row in df.iterrows(): flag, adj, calc, corrected, reason = reconcile_record(row) results.append({ "donor_id": row["donor_id"], "name": row["name"], "manual_points": row["manual_priority_points"], "calculated_points": calc, "adjustment_delta": adj, "corrected_points": corrected, "flag_type": flag, "flag_reason": reason }) results_df = pd.DataFrame(results) results_df.to_csv("changes_preview.csv", index=False) print(f"Preview generated: {len(results_df)} records evaluated.") if args.dry_run: print("DRY RUN COMPLETE. No API calls made.") return errors_df = results_df[results_df["flag_type"] == "ERROR"] warnings_count = len(results_df[results_df["flag_type"] == "WARNING"]) print(f"Pushing {len(errors_df)} ERROR records to Spectra API...") successes, failures = process_api_updates(errors_df, config) print(f"Run Summary: {successes} updated, {warnings_count} warnings skipped, {failures} failed. See {config['db_path']} for audit log.") if name == "__main__": main() Rollback Utility (rollback_utility.py) If a business rule was misapplied, this script reads the SQLite audit log and safely reverts the API back to the spectra_old_value recorded immediately prior to the ETL run. import argparse import sqlite3 from priority_points_etl import load_config, api_request def rollback(run_date_prefix): config = load_config() headers = {"Authorization": f"Bearer {config['auth_token']}", "Content-Type": "application/json"} conn = sqlite3.connect(config["db_path"]) cur = conn.cursor() # Fetch successful updates from a specific date/time prefix (e.g., "2024-10-25") cur.execute(""" SELECT donor_id, spectra_old_value FROM run_log WHERE success = 1 AND timestamp LIKE ? AND spectra_old_value IS NOT NULL """, (f"{run_date_prefix}%",)) records = cur.fetchall() print(f"Found {len(records)} records to rollback for prefix '{run_date_prefix}'.") success_count = 0 for donor_id, old_value in records: url = config["api_base_url"] + config["endpoint_template"].format(donor_id=donor_id) resp = api_request("PATCH", url, headers, payload={"priorityPoints": old_value}) if resp and 200 <= resp.status_code < 300: success_count += 1 cur.execute(""" INSERT INTO run_log (donor_id, timestamp, spectra_old_value, new_value, api_status_code, success) VALUES (?, datetime('now'), ?, ?, ?, 1) """, (donor_id, old_value, old_value, resp.status_code)) conn.commit() print(f"Rollback complete: {success_count}/{len(records)} successfully reverted.") conn.close() if name == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--date", required=True, help="Timestamp prefix to rollback (e.g., '2024-10-25')") args = parser.parse_args() rollback(args.date) Data Mapping Plan (data_mapping_plan.md) # Army A Club Priority Points ETL — Data Mapping & Go-Live Plan Priority points directly affect donor access and benefits (seating, parking, post-season access). Because this data is highly sensitive, the process uses a reconcile-first, preview, and verified-load pattern with a strict SQLite rollback log. Pre-Go-Live Verification Checklist (CRITICAL) [ ] Confirm Business Rules: The 2025 Army A Club Benefits Chart states priority is determined first by membership level, then by points. The script's formula (1 pt / $100 + tenure multiplier) is a placeholder and MUST be explicitly approved by USMA AOG leadership. [ ] Verify Spectra API Endpoints: Confirm with Paciolan support that PATCH /api/v1/donors/{donor_id}/priority-points is the correct endpoint for the USMA tenant, and that priorityPoints is the exact JSON payload key. [ ] Environment Variables: Ensure .env is configured on the deployment machine and added to .gitignore. Never commit SPECTRA_AUTH_TOKEN. Field Mapping Legacy / CSV Field Spectra API Field Type Transformation / Notes donor_id DONOR_ID (URL Path) String Primary rollback key in SQLite log. name N/A String Used in local preview only; masked from API logs for privacy. manual_priority_points N/A Integer Legacy value used to trigger ERROR flags. adjustment_notes N/A String Parsed via Regex to extract signed integer bonuses. donation_history N/A JSON Used internally to calculate ground-truth points. (From GET Request) priorityPoints Integer Captured as spectra_old_value in the SQLite log for rollback. (Calculated) priorityPoints Integer Pushed via PATCH only if the record is flagged as ERROR. Reconciliation Rules Condition Flag Action manual_points == calculated_points + adjustment_amount CLEAN No API action. Mismatch + text note exists, but no numeric amount provided WARNING No API action; human review required. Mismatch + no explanatory note or amount ERROR Eligible for API update. Mismatch + note/amount exists, but math still does not reconcile ERROR Eligible for API update. Privacy / Security Notes PCI Compliance: Do not process SSNs, card data, or other PCI data in this ETL. Data in Transit: Use HTTPS and Bearer token authentication for all API calls. PII Handling: Mask names in operational logs; the SQLite rollback log stores donor_id only. Mermaid ETL Flowchart Sources [1] 5 Best Practices for Effective Donor Database for Colleges — https://empowersis.com/donor-database-colleges/ [3] 10 Best Practices For Effective Donor Data Management - Affnetz — https://affnetz.com/10-best-practices-for-effective-donor-data-management/ [4] 2025 benefits chart - West Point Athletics — https://goarmywestpoint.com/documents/download/2024/3/18/2025_Army_A_Club_Benefits_Chart.pdf [6] 4 Best Practices for Higher Ed Fundraising Software Success — https://www.classter.com/blog/future-of-learning-management-systems/4-best-practices-for-higher-ed-fundraising-software-success/