Implementation Guide: Identify major gift prospects based on giving history and engagement signals

Install-Module -Name SKYAPIPowerShell -Scope CurrentUser
Connect-SKYAPI -ClientId 'YOUR_CLIENT_ID' -ClientSecret 'YOUR_CLIENT_SECRET' -RedirectUri 'https://localhost'
$donors = Get-SKYAPIConstituent -Limit 10000
$donors | Export-Csv -Path 'C:\DataAudit\donor_export.csv' -NoTypeInformation

java -cp dataloader.jar -Dsalesforce.config.dir=conf process csvExtractDonors

pip install pandas numpy
python data_quality_audit.py --input donor_export.csv --output audit_report.html

pip install requests pandas
python validate_emails.py --input donor_export.csv --api_key 'YOUR_NEVERBOUNCE_KEY' --output validated_emails.csv

#!/usr/bin/env python3
"""Donor Data Quality Audit Script for AI Prospect Identification Projects
Usage: python data_quality_audit.py --input donor_export.csv --output audit_report.html
"""
import pandas as pd
import numpy as np
import argparse
import re
from datetime import datetime, timedelta
from collections import Counter

def load_data(filepath):
    df = pd.read_csv(filepath, low_memory=False, encoding='utf-8-sig')
    df.columns = [c.strip().lower().replace(' ', '_') for c in df.columns]
    return df

def detect_column_mapping(df):
    """Auto-detect common CRM column names and map to standard fields."""
    mappings = {
        'donor_id': ['constituent_id', 'donor_id', 'id', 'contact_id', 'account_number', 'record_id'],
        'first_name': ['first_name', 'first', 'fname', 'given_name'],
        'last_name': ['last_name', 'last', 'lname', 'surname', 'family_name'],
        'email': ['email', 'email_address', 'primary_email', 'e-mail'],
        'phone': ['phone', 'phone_number', 'primary_phone', 'home_phone', 'mobile'],
        'address': ['address', 'street', 'address_line_1', 'mailing_address', 'street_address'],
        'city': ['city', 'mailing_city'],
        'state': ['state', 'st', 'mailing_state', 'province'],
        'zip': ['zip', 'zip_code', 'postal_code', 'zipcode', 'mailing_zip'],
        'total_giving': ['total_giving', 'total_gifts', 'lifetime_giving', 'total_donated', 'cumulative_giving'],
        'last_gift_date': ['last_gift_date', 'last_donation_date', 'most_recent_gift', 'latest_gift_date'],
        'last_gift_amount': ['last_gift_amount', 'last_donation_amount', 'most_recent_gift_amount'],
        'first_gift_date': ['first_gift_date', 'first_donation_date', 'earliest_gift_date'],
        'donor_status': ['status', 'donor_status', 'constituent_status', 'record_status'],
        'deceased': ['deceased', 'is_deceased', 'deceased_flag'],
        'do_not_contact': ['do_not_contact', 'dnc', 'solicit_flag', 'no_contact']
    }
    result = {}
    for standard, options in mappings.items():
        for opt in options:
            if opt in df.columns:
                result[standard] = opt
                break
    return result

def analyze_completeness(df, col_map):
    results = []
    for field, col in col_map.items():
        total = len(df)
        non_null = df[col].notna().sum()
        non_empty = df[col].astype(str).str.strip().replace('', np.nan).notna().sum()
        pct_complete = round((non_empty / total) * 100, 1)
        results.append({'Field': field, 'Column': col, 'Total_Records': total,
                        'Populated': non_empty, 'Missing': total - non_empty,
                        'Completeness_%': pct_complete,
                        'Status': 'GOOD' if pct_complete >= 90 else 'WARNING' if pct_complete >= 70 else 'CRITICAL'})
    return pd.DataFrame(results)

def detect_duplicates(df, col_map):
    dupe_checks = []
    if 'first_name' in col_map and 'last_name' in col_map:
        name_col = [col_map['first_name'], col_map['last_name']]
        if 'zip' in col_map:
            name_col.append(col_map['zip'])
        dupes = df[df.duplicated(subset=name_col, keep=False)]
        dupe_checks.append({'Method': 'Name + ZIP', 'Duplicate_Records': len(dupes),
                            'Duplicate_Groups': len(dupes.drop_duplicates(subset=name_col)),
                            'Pct_of_Total': round((len(dupes)/len(df))*100, 1)})
    if 'email' in col_map:
        email_col = col_map['email']
        email_dupes = df[df[email_col].notna() & df.duplicated(subset=[email_col], keep=False)]
        dupe_checks.append({'Method': 'Email Address', 'Duplicate_Records': len(email_dupes),
                            'Duplicate_Groups': len(email_dupes.drop_duplicates(subset=[email_col])),
                            'Pct_of_Total': round((len(email_dupes)/len(df))*100, 1)})
    return pd.DataFrame(dupe_checks)

def validate_emails(df, col_map):
    if 'email' not in col_map:
        return {'valid': 0, 'invalid': 0, 'missing': len(df)}
    email_col = col_map['email']
    emails = df[email_col].dropna().astype(str)
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    valid = emails.str.match(pattern).sum()
    invalid = len(emails) - valid
    missing = len(df) - len(emails)
    return {'valid': int(valid), 'invalid': int(invalid), 'missing': int(missing)}

def analyze_giving_history(df, col_map):
    results = {}
    if 'last_gift_date' in col_map:
        dates = pd.to_datetime(df[col_map['last_gift_date']], errors='coerce')
        now = datetime.now()
        results['donors_gave_last_12mo'] = int((dates > now - timedelta(days=365)).sum())
        results['donors_gave_last_36mo'] = int((dates > now - timedelta(days=1095)).sum())
        results['donors_lapsed_3plus_years'] = int((dates <= now - timedelta(days=1095)).sum())
        results['earliest_gift_on_file'] = str(dates.min().date()) if dates.notna().any() else 'N/A'
        results['years_of_history'] = round((now - dates.min()).days / 365.25, 1) if dates.notna().any() else 0
    if 'total_giving' in col_map:
        giving = pd.to_numeric(df[col_map['total_giving']], errors='coerce')
        results['total_donors_with_gifts'] = int(giving[giving > 0].count())
        results['median_lifetime_giving'] = round(float(giving[giving > 0].median()), 2)
        results['mean_lifetime_giving'] = round(float(giving[giving > 0].mean()), 2)
        results['donors_above_1000'] = int((giving >= 1000).sum())
        results['donors_above_5000'] = int((giving >= 5000).sum())
        results['donors_above_10000'] = int((giving >= 10000).sum())
    return results

def ai_readiness_score(completeness_df, dupe_df, giving_stats, total_records):
    score = 100
    deductions = []
    # Record count
    if total_records < 500:
        score -= 30
        deductions.append('CRITICAL: Fewer than 500 records (minimum for DonorSearch)')
    elif total_records < 10000:
        score -= 10
        deductions.append('WARNING: Fewer than 10,000 records (Dataro recommends 10K+ for best results)')
    # Data completeness
    for _, row in completeness_df.iterrows():
        if row['Status'] == 'CRITICAL':
            score -= 10
            deductions.append(f"CRITICAL: {row['Field']} only {row['Completeness_%']}% complete")
        elif row['Status'] == 'WARNING':
            score -= 5
            deductions.append(f"WARNING: {row['Field']} only {row['Completeness_%']}% complete")
    # Duplicates
    if not dupe_df.empty and dupe_df['Pct_of_Total'].max() > 15:
        score -= 15
        deductions.append(f"CRITICAL: {dupe_df['Pct_of_Total'].max()}% duplicate records detected")
    elif not dupe_df.empty and dupe_df['Pct_of_Total'].max() > 5:
        score -= 5
        deductions.append(f"WARNING: {dupe_df['Pct_of_Total'].max()}% duplicate records detected")
    # Giving history depth
    years = giving_stats.get('years_of_history', 0)
    if years < 3:
        score -= 20
        deductions.append(f'CRITICAL: Only {years} years of giving history (3-5 years recommended)')
    elif years < 5:
        score -= 5
        deductions.append(f'WARNING: Only {years} years of giving history (5+ years ideal for Dataro)')
    return max(0, min(100, score)), deductions

def generate_html_report(completeness_df, dupe_df, email_stats, giving_stats, readiness_score, deductions, total_records):
    readiness_color = '#27ae60' if readiness_score >= 80 else '#f39c12' if readiness_score >= 60 else '#e74c3c'
    html = f"""<!DOCTYPE html><html><head><title>Donor Data Quality Audit Report</title>
    <style>body{{font-family:Arial,sans-serif;margin:40px;}}table{{border-collapse:collapse;width:100%;margin:20px 0;}}
    th,td{{border:1px solid #ddd;padding:8px;text-align:left;}}th{{background:#2c3e50;color:white;}}
    .good{{color:#27ae60;font-weight:bold;}}.warning{{color:#f39c12;font-weight:bold;}}.critical{{color:#e74c3c;font-weight:bold;}}
    .score-box{{font-size:48px;color:{readiness_color};font-weight:bold;text-align:center;padding:20px;border:3px solid {readiness_color};border-radius:10px;width:200px;margin:20px auto;}}
    h1{{color:#2c3e50;}}h2{{color:#34495e;border-bottom:2px solid #3498db;padding-bottom:5px;}}</style></head><body>
    <h1>Donor Data Quality Audit Report</h1>
    <p>Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
    <p>Total Records Analyzed: <strong>{total_records:,}</strong></p>
    <h2>AI Readiness Score</h2>
    <div class='score-box'>{readiness_score}/100</div>
    <ul>{''.join(f'<li class="{"critical" if "CRITICAL" in d else "warning"}">{d}</li>' for d in deductions)}</ul>
    <h2>Field Completeness</h2>{completeness_df.to_html(index=False, classes='completeness')}
    <h2>Duplicate Analysis</h2>{dupe_df.to_html(index=False) if not dupe_df.empty else '<p>No duplicate analysis possible with available columns.</p>'}
    <h2>Email Validation</h2><ul><li>Valid: {email_stats['valid']:,}</li><li>Invalid format: {email_stats['invalid']:,}</li><li>Missing: {email_stats['missing']:,}</li></ul>
    <h2>Giving History Analysis</h2><ul>{''.join(f'<li><strong>{k.replace("_"," ").title()}</strong>: {v}</li>' for k,v in giving_stats.items())}</ul>
    <h2>Recommendations</h2><ol>
    <li>{'De-duplicate records before AI platform ingestion' if not dupe_df.empty and dupe_df['Pct_of_Total'].max() > 5 else 'Duplicate rate is acceptable'}</li>
    <li>{'Validate and update email addresses using NeverBounce or ZeroBounce' if email_stats['invalid'] > 0 else 'Email data looks clean'}</li>
    <li>{'Run NCOA address update to ensure mailing addresses are current' if completeness_df[completeness_df['Field']=="address"]['Completeness_%'].values[0] < 90 if 'address' in completeness_df['Field'].values else True else 'Address data looks sufficient'}</li>
    <li>Recommended AI Platform: {'DonorSearch Enhanced CORE (500-9,999 records)' if total_records < 10000 else 'Dataro or DonorSearch AI Full (10,000+ records)'}</li>
    </ol></body></html>"""
    return html

def main():
    parser = argparse.ArgumentParser(description='Donor Data Quality Audit for AI Prospect Identification')
    parser.add_argument('--input', required=True, help='Path to donor CSV export')
    parser.add_argument('--output', default='audit_report.html', help='Output HTML report path')
    args = parser.parse_args()
    
    print(f'Loading data from {args.input}...')
    df = load_data(args.input)
    print(f'Loaded {len(df):,} records with {len(df.columns)} columns')
    
    col_map = detect_column_mapping(df)
    print(f'Detected column mappings: {col_map}')
    
    completeness = analyze_completeness(df, col_map)
    duplicates = detect_duplicates(df, col_map)
    emails = validate_emails(df, col_map)
    giving = analyze_giving_history(df, col_map)
    score, deductions = ai_readiness_score(completeness, duplicates, giving, len(df))
    
    html = generate_html_report(completeness, duplicates, emails, giving, score, deductions, len(df))
    with open(args.output, 'w') as f:
        f.write(html)
    print(f'Report generated: {args.output}')
    print(f'AI Readiness Score: {score}/100')

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""CRM-to-AI Platform Sync Health Monitor
Schedule via cron: 0 7 * * * /usr/bin/python3 /opt/msp/sync_monitor.py
Or run as Azure Function with Timer Trigger.
"""
import requests
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta
import os

# Configuration - store in environment variables or Azure Key Vault
CONFIG = {
    'crm_type': os.getenv('CRM_TYPE', 'raiser_edge_nxt'),  # raiser_edge_nxt | salesforce | donorperfect
    'crm_api_url': os.getenv('CRM_API_URL', 'https://api.sky.blackbaud.com/constituent/v1'),
    'crm_api_key': os.getenv('CRM_API_KEY'),
    'crm_oauth_token': os.getenv('CRM_OAUTH_TOKEN'),
    'ai_platform': os.getenv('AI_PLATFORM', 'donorsearch'),  # donorsearch | dataro
    'ai_api_url': os.getenv('AI_API_URL'),
    'ai_api_key': os.getenv('AI_API_KEY'),
    'smtp_server': os.getenv('SMTP_SERVER', 'smtp.office365.com'),
    'smtp_port': int(os.getenv('SMTP_PORT', '587')),
    'smtp_user': os.getenv('SMTP_USER'),
    'smtp_pass': os.getenv('SMTP_PASS'),
    'alert_recipients': os.getenv('ALERT_RECIPIENTS', 'msp-alerts@yourmsp.com').split(','),
    'client_name': os.getenv('CLIENT_NAME', 'Nonprofit Client'),
    'sync_freshness_hours': int(os.getenv('SYNC_FRESHNESS_HOURS', '48')),
    'record_count_tolerance_pct': float(os.getenv('RECORD_TOLERANCE_PCT', '2.0')),
}

def get_crm_record_count():
    """Get active donor count from CRM."""
    if CONFIG['crm_type'] == 'raiser_edge_nxt':
        headers = {
            'Bb-Api-Subscription-Key': CONFIG['crm_api_key'],
            'Authorization': f"Bearer {CONFIG['crm_oauth_token']}"
        }
        resp = requests.get(f"{CONFIG['crm_api_url']}/constituents?limit=1", headers=headers)
        return resp.json().get('count', 0)
    elif CONFIG['crm_type'] == 'salesforce':
        headers = {'Authorization': f"Bearer {CONFIG['crm_oauth_token']}"}
        query = 'SELECT COUNT() FROM Contact WHERE npe01__Donor__c = true'
        resp = requests.get(f"{CONFIG['crm_api_url']}/query?q={query}", headers=headers)
        return resp.json().get('totalSize', 0)
    elif CONFIG['crm_type'] == 'donorperfect':
        headers = {'Authorization': f"Bearer {CONFIG['crm_api_key']}"}
        resp = requests.get(f"{CONFIG['crm_api_url']}/donors?count_only=true", headers=headers)
        return resp.json().get('count', 0)
    return 0

def get_ai_platform_status():
    """Get record count and last sync timestamp from AI platform."""
    headers = {'Authorization': f"Bearer {CONFIG['ai_api_key']}"}
    if CONFIG['ai_platform'] == 'donorsearch':
        resp = requests.get(f"{CONFIG['ai_api_url']}/account/sync-status", headers=headers)
        data = resp.json()
        return {'record_count': data.get('total_records', 0),
                'last_sync': data.get('last_sync_timestamp', ''),
                'sync_status': data.get('status', 'unknown')}
    elif CONFIG['ai_platform'] == 'dataro':
        resp = requests.get(f"{CONFIG['ai_api_url']}/v1/integration/status", headers=headers)
        data = resp.json()
        return {'record_count': data.get('synced_contacts', 0),
                'last_sync': data.get('last_prediction_run', ''),
                'sync_status': data.get('integration_status', 'unknown')}
    return {'record_count': 0, 'last_sync': '', 'sync_status': 'unknown'}

def check_sync_health():
    issues = []
    crm_count = get_crm_record_count()
    ai_status = get_ai_platform_status()
    ai_count = ai_status['record_count']
    
    # Check record count discrepancy
    if crm_count > 0 and ai_count > 0:
        discrepancy = abs(crm_count - ai_count) / crm_count * 100
        if discrepancy > CONFIG['record_count_tolerance_pct']:
            issues.append(f"RECORD COUNT MISMATCH: CRM has {crm_count:,} records, AI platform has {ai_count:,} ({discrepancy:.1f}% discrepancy, threshold is {CONFIG['record_count_tolerance_pct']}%)")
    elif ai_count == 0:
        issues.append(f"CRITICAL: AI platform reports 0 records. CRM has {crm_count:,}. Sync may be broken.")
    
    # Check sync freshness
    if ai_status['last_sync']:
        last_sync = datetime.fromisoformat(ai_status['last_sync'].replace('Z', '+00:00'))
        hours_since = (datetime.now(last_sync.tzinfo) - last_sync).total_seconds() / 3600
        if hours_since > CONFIG['sync_freshness_hours']:
            issues.append(f"STALE DATA: Last sync was {hours_since:.0f} hours ago (threshold: {CONFIG['sync_freshness_hours']} hours). Last sync: {last_sync.strftime('%Y-%m-%d %H:%M')}")
    else:
        issues.append('WARNING: Unable to determine last sync timestamp.')
    
    # Check sync status
    if ai_status['sync_status'] not in ['active', 'healthy', 'connected', 'ok']:
        issues.append(f"SYNC STATUS ISSUE: AI platform reports sync status as '{ai_status['sync_status']}'")
    
    return {'healthy': len(issues) == 0, 'issues': issues, 'crm_count': crm_count,
            'ai_count': ai_count, 'last_sync': ai_status['last_sync'], 'sync_status': ai_status['sync_status']}

def send_alert(health_report):
    msg = MIMEMultipart('alternative')
    msg['Subject'] = f"[{'ALERT' if not health_report['healthy'] else 'OK'}] AI Sync Monitor - {CONFIG['client_name']}"
    msg['From'] = CONFIG['smtp_user']
    msg['To'] = ', '.join(CONFIG['alert_recipients'])
    body = f"""AI Prospect Platform Sync Health Report\n{'='*50}\nClient: {CONFIG['client_name']}\nTimestamp: {datetime.now().strftime('%Y-%m-%d %H:%M')}\nCRM Type: {CONFIG['crm_type']}\nAI Platform: {CONFIG['ai_platform']}\nCRM Record Count: {health_report['crm_count']:,}\nAI Platform Record Count: {health_report['ai_count']:,}\nLast Sync: {health_report['last_sync']}\nSync Status: {health_report['sync_status']}\nOverall Health: {'HEALTHY' if health_report['healthy'] else 'ISSUES DETECTED'}\n\n"""
    if health_report['issues']:
        body += 'Issues Detected:\n' + '\n'.join(f'  - {i}' for i in health_report['issues'])
        body += '\n\nAction Required: Review integration settings and re-sync if necessary.'
    msg.attach(MIMEText(body, 'plain'))
    with smtplib.SMTP(CONFIG['smtp_server'], CONFIG['smtp_port']) as server:
        server.starttls()
        server.login(CONFIG['smtp_user'], CONFIG['smtp_pass'])
        server.sendmail(CONFIG['smtp_user'], CONFIG['alert_recipients'], msg.as_string())

def main():
    try:
        health = check_sync_health()
        if not health['healthy']:
            send_alert(health)
            print(f"ALERT sent: {len(health['issues'])} issues detected")
        else:
            print(f"Sync healthy: CRM={health['crm_count']:,}, AI={health['ai_count']:,}")
            # Send weekly OK report on Mondays
            if datetime.now().weekday() == 0:
                send_alert(health)
    except Exception as e:
        error_report = {'healthy': False, 'issues': [f'MONITOR ERROR: {str(e)}'], 'crm_count': 0, 'ai_count': 0, 'last_sync': 'unknown', 'sync_status': 'monitor_error'}
        send_alert(error_report)

if __name__ == '__main__':
    main()

# pulls AI prospect scores and updates SharePoint Prospect_Scores_History
# list. Run before the Power Automate flow triggers at 7:30 AM.

import requests
import json
from datetime import datetime

def refresh_prospect_scores(ai_api_url, ai_api_key, sharepoint_site, sp_list_id, sp_access_token):
    """Pull latest scores from AI platform and update SharePoint tracking list."""
    # Get current scores from AI platform
    headers = {'Authorization': f'Bearer {ai_api_key}'}
    resp = requests.get(f'{ai_api_url}/prospects/scores', headers=headers)
    current_scores = {p['donor_id']: p for p in resp.json()['prospects']}
    
    # Get previous scores from SharePoint
    sp_headers = {'Authorization': f'Bearer {sp_access_token}',
                  'Content-Type': 'application/json'}
    sp_url = f'https://graph.microsoft.com/v1.0/sites/{sharepoint_site}/lists/{sp_list_id}/items?$expand=fields'
    sp_resp = requests.get(sp_url, headers=sp_headers)
    existing = {item['fields']['donor_id']: item for item in sp_resp.json().get('value', [])}
    
    def get_tier(score):
        if score >= 80: return 'Tier 1'
        elif score >= 60: return 'Tier 2'
        elif score >= 40: return 'Tier 3'
        return 'Tier 4'
    
    changes = []
    for donor_id, data in current_scores.items():
        new_score = data.get('propensity_score', 0)
        new_tier = get_tier(new_score)
        old_score = existing.get(donor_id, {}).get('fields', {}).get('current_score', 0)
        old_tier = get_tier(old_score) if old_score else 'Tier 4'
        
        if new_tier != old_tier:
            update_payload = {'fields': {
                'donor_id': donor_id,
                'donor_name': data.get('name', ''),
                'previous_score': old_score,
                'current_score': new_score,
                'previous_tier': old_tier,
                'current_tier': new_tier,
                'score_change_date': datetime.utcnow().isoformat(),
                'capacity_estimate': data.get('estimated_capacity', 0),
                'last_gift_amount': data.get('last_gift_amount', 0),
                'alert_sent': False
            }}
            if donor_id in existing:
                item_id = existing[donor_id]['id']
                requests.patch(f'{sp_url}/{item_id}', headers=sp_headers, json=update_payload)
            else:
                requests.post(f'{sp_url}', headers=sp_headers, json=update_payload)
            changes.append(f"{data.get('name')}: {old_tier} -> {new_tier} (Score: {old_score} -> {new_score})")
    
    return changes

# paste into the Model view Measures table

Total Prospects = COUNTROWS(Prospects)

Tier 1 Count = CALCULATE(COUNTROWS(Prospects), Prospects[ScoreTier] = "Tier 1")

Tier 2 Count = CALCULATE(COUNTROWS(Prospects), Prospects[ScoreTier] = "Tier 2")

Total Pipeline Value = SUMX(
    FILTER(Prospects, Prospects[AIScore] >= 60),
    Prospects[RecommendedAskAmount]
)

Avg Days Since Contact = AVERAGEX(
    FILTER(Prospects, Prospects[AIScore] >= 60),
    DATEDIFF(Prospects[LastContactDate], TODAY(), DAY)
)

Prospects Needing Contact = CALCULATE(
    COUNTROWS(Prospects),
    Prospects[AIScore] >= 60,
    DATEDIFF(Prospects[LastContactDate], TODAY(), DAY) > 30
)

New Tier 1 This Month = CALCULATE(
    COUNTROWS(ScoreHistory),
    ScoreHistory[Tier] = "Tier 1",
    MONTH(ScoreHistory[ScoreDate]) = MONTH(TODAY()),
    YEAR(ScoreHistory[ScoreDate]) = YEAR(TODAY())
)

Conversion Rate = DIVIDE(
    CALCULATE(COUNTROWS(Prospects), Prospects[LifetimeGiving] >= [MajorGiftThreshold], Prospects[AIScore] >= 60),
    CALCULATE(COUNTROWS(Prospects), Prospects[AIScore] >= 60),
    0
)

Engagement Depth = AVERAGEX(
    Prospects,
    CALCULATE(
        COUNTROWS(EngagementSignals),
        FILTER(EngagementSignals, EngagementSignals[EngagementDate] >= TODAY() - 365)
    )
)

Hidden Gems = CALCULATE(
    COUNTROWS(Prospects),
    CONTAINSSTRING(Prospects[Tags], "Hidden Gem")
)

# calculates a 0–100 engagement score based on recency, frequency, and
# breadth of donor interactions

#!/usr/bin/env python3
"""Donor Engagement Score Calculator
Calculates a 0-100 engagement score based on recency, frequency, and breadth
of donor interactions. Designed to supplement AI prospect propensity scores.

Usage:
  python engagement_scorer.py --input engagements.csv --donors donors.csv --output scored_donors.csv

Input CSV format (engagements.csv):
  donor_id, engagement_type, engagement_date, engagement_value

Engagement types: gift, email_open, email_click, event_attended, volunteer_hours,
                  meeting, phone_call, website_visit, social_media, survey_response,
                  board_service, committee_service, peer_referral
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import argparse

# Engagement type weights (configurable per client)
ENGAGEMENT_WEIGHTS = {
    'gift': 10.0,
    'meeting': 8.0,
    'board_service': 9.0,
    'committee_service': 7.0,
    'event_attended': 6.0,
    'volunteer_hours': 5.0,
    'peer_referral': 7.0,
    'phone_call': 5.0,
    'email_click': 3.0,
    'survey_response': 4.0,
    'email_open': 1.5,
    'website_visit': 1.0,
    'social_media': 1.0,
}

# Recency decay: more recent = higher multiplier
def recency_multiplier(days_ago):
    """Exponential decay: interactions today = 1.0x, 1 year ago = 0.5x, 2 years ago = 0.25x"""
    half_life = 365  # days
    return np.power(0.5, days_ago / half_life)

def calculate_engagement_scores(engagements_df, donors_df):
    now = datetime.now()
    engagements_df['engagement_date'] = pd.to_datetime(engagements_df['engagement_date'], errors='coerce')
    engagements_df['days_ago'] = (now - engagements_df['engagement_date']).dt.days
    engagements_df['days_ago'] = engagements_df['days_ago'].clip(lower=0)
    
    # Apply weights and recency decay
    engagements_df['base_weight'] = engagements_df['engagement_type'].map(ENGAGEMENT_WEIGHTS).fillna(1.0)
    engagements_df['recency_mult'] = engagements_df['days_ago'].apply(recency_multiplier)
    engagements_df['weighted_score'] = engagements_df['base_weight'] * engagements_df['recency_mult']
    
    # Aggregate per donor
    donor_scores = engagements_df.groupby('donor_id').agg(
        raw_score=('weighted_score', 'sum'),
        total_interactions=('engagement_type', 'count'),
        unique_types=('engagement_type', 'nunique'),
        most_recent=('engagement_date', 'max'),
        gift_count=('engagement_type', lambda x: (x == 'gift').sum()),
        last_gift_days=('days_ago', lambda x: x[engagements_df.loc[x.index, 'engagement_type'] == 'gift'].min() if (engagements_df.loc[x.index, 'engagement_type'] == 'gift').any() else 9999),
    ).reset_index()
    
    # Breadth bonus: engaging across multiple channels is a strong signal
    max_types = len(ENGAGEMENT_WEIGHTS)
    donor_scores['breadth_bonus'] = (donor_scores['unique_types'] / max_types) * 15
    
    # Frequency bonus: consistent engagement matters
    # Calculate interactions per month over last 24 months
    recent = engagements_df[engagements_df['days_ago'] <= 730]
    monthly_freq = recent.groupby('donor_id').agg(
        months_active=('engagement_date', lambda x: x.dt.to_period('M').nunique())
    ).reset_index()
    donor_scores = donor_scores.merge(monthly_freq, on='donor_id', how='left')
    donor_scores['months_active'] = donor_scores['months_active'].fillna(0)
    donor_scores['frequency_bonus'] = (donor_scores['months_active'] / 24) * 10
    
    # Final score: raw + breadth + frequency, normalized to 0-100
    donor_scores['total_raw'] = donor_scores['raw_score'] + donor_scores['breadth_bonus'] + donor_scores['frequency_bonus']
    
    # Normalize using percentile ranking (0-100 scale)
    donor_scores['engagement_score'] = donor_scores['total_raw'].rank(pct=True) * 100
    donor_scores['engagement_score'] = donor_scores['engagement_score'].round(0).astype(int)
    
    # Engagement tier
    donor_scores['engagement_tier'] = pd.cut(
        donor_scores['engagement_score'],
        bins=[0, 25, 50, 75, 100],
        labels=['Low', 'Moderate', 'High', 'Very High'],
        include_lowest=True
    )
    
    # Merge with donor master
    result = donors_df.merge(
        donor_scores[['donor_id', 'engagement_score', 'engagement_tier', 'total_interactions',
                       'unique_types', 'most_recent', 'months_active']],
        on='donor_id', how='left'
    )
    result['engagement_score'] = result['engagement_score'].fillna(0).astype(int)
    result['engagement_tier'] = result['engagement_tier'].fillna('None')
    
    return result

def main():
    parser = argparse.ArgumentParser(description='Calculate donor engagement scores')
    parser.add_argument('--input', required=True, help='Engagements CSV file')
    parser.add_argument('--donors', required=True, help='Donors CSV file')
    parser.add_argument('--output', default='scored_donors.csv', help='Output CSV')
    parser.add_argument('--weights-config', default=None, help='Optional JSON file to override engagement weights')
    args = parser.parse_args()
    
    if args.weights_config:
        import json
        with open(args.weights_config) as f:
            custom_weights = json.load(f)
            ENGAGEMENT_WEIGHTS.update(custom_weights)
            print(f'Loaded custom weights from {args.weights_config}')
    
    engagements = pd.read_csv(args.input)
    donors = pd.read_csv(args.donors)
    
    print(f'Processing {len(engagements):,} engagement records for {len(donors):,} donors...')
    result = calculate_engagement_scores(engagements, donors)
    
    result.to_csv(args.output, index=False)
    print(f'Scored donors exported to {args.output}')
    print(f"\nEngagement Score Distribution:")
    print(result['engagement_tier'].value_counts().sort_index())
    print(f"\nTop 10 Most Engaged Donors:")
    print(result.nlargest(10, 'engagement_score')[['donor_id', 'engagement_score', 'engagement_tier', 'total_interactions', 'unique_types']].to_string(index=False))

if __name__ == '__main__':
    main()

# AI Governance Policy Generator Prompt Template
# Use with: ChatGPT-4, Claude, or Copilot
# Replace all {PLACEHOLDERS} with client-specific information before running

---
SYSTEM PROMPT:
You are a nonprofit governance and data privacy expert. Generate a comprehensive
AI Acceptable Use Policy for a 501(c)(3) nonprofit organization. The policy must
be practical, board-ready, and compliant with applicable state privacy laws.
Write in clear, non-technical language appropriate for nonprofit board members
and development staff.

USER PROMPT:
Please generate an AI Acceptable Use Policy for our nonprofit organization with
the following details:

- Organization Name: {ORGANIZATION_NAME}
- EIN: {EIN}
- State of Incorporation: {STATE}
- Operating States: {OPERATING_STATES_LIST}
- Annual Revenue: {ANNUAL_REVENUE}
- Number of Donor Records: {RECORD_COUNT}
- AI Tools in Use: {AI_TOOLS_LIST} (e.g., DonorSearch AI, Dataro, Power BI)
- CRM Platform: {CRM_PLATFORM}
- Staff Size: {STAFF_SIZE}
- Has existing data privacy policy: {YES_NO}
- Processes donor credit card data: {YES_NO}
- Has international (EU/UK) donors: {YES_NO}

The policy must include these sections:

1. PURPOSE AND SCOPE
   - Why the organization uses AI tools
   - Which staff and systems are covered
   - Relationship to existing data governance policies

2. APPROVED AI TOOLS AND USES
   - Enumerated list of approved AI platforms and their purposes
   - Approved use cases (prospect identification, donor segmentation, lapse prediction)
   - Prohibited uses (donor score sharing outside org, automated solicitation without human review, using AI scores for employment or volunteer decisions)

3. DONOR DATA HANDLING
   - What donor data is shared with AI platforms
   - Data minimization principles
   - Vendor data processing agreements required
   - Data retention and deletion schedules
   - Donor opt-out rights and process

4. TRANSPARENCY AND DISCLOSURE
   - Language to add to privacy policy regarding AI use
   - When and how to disclose AI involvement to donors
   - Board reporting requirements on AI tool usage

5. BIAS AND FAIRNESS
   - Quarterly review of AI-generated prospect lists for demographic bias
   - Process for investigating and addressing scoring anomalies
   - Prohibition on using AI scores as sole basis for donor cultivation decisions

6. STATE PRIVACY LAW COMPLIANCE
   - Specific requirements for {OPERATING_STATES_LIST}
   - Special attention to Colorado Consumer Privacy Act (if applicable)
   - Special attention to Oregon Consumer Privacy Act (if applicable, effective July 1, 2025 for nonprofits)
   - Consent requirements and opt-out mechanisms

7. SECURITY REQUIREMENTS
   - MFA required for all AI platform access
   - Role-based access controls
   - Annual security review of AI vendor SOC 2 compliance
   - Incident response for AI-related data breaches

8. STAFF RESPONSIBILITIES
   - Required training before AI tool access
   - Restrictions on entering donor data into general-purpose AI (ChatGPT, etc.)
   - Reporting obligations for suspected misuse

9. GOVERNANCE AND REVIEW
   - Policy review schedule (annual minimum)
   - Responsible officer (typically Director of Development or COO)
   - Board oversight requirements
   - Policy amendment process

Format as a professional policy document with section numbers, effective date placeholder, and signature lines for Board Chair and Executive Director.