""" CSV Import Templates Module This module provides standardized CSV templates for importing data into the membership platform. Templates are designed to be source-agnostic, allowing clients to transform their data from any system to match our schema. Supported Templates: - users.csv - Core user data - subscriptions.csv - Membership/plan history - donations.csv - Donation history - payments.csv - Payment transactions - registration_data.csv - Custom registration fields Author: Claude Code Date: 2026-02-03 """ import csv import io from typing import Dict, List, Any, Optional, Tuple from datetime import datetime, date import re import logging logger = logging.getLogger(__name__) # ============================================================================= # Template Definitions # ============================================================================= TEMPLATES = { "users": { "name": "Users", "description": "Core user data including personal information, contact details, and account settings", "required": True, "filename": "users_template.csv", "columns": [ # Required fields {"name": "email", "required": True, "type": "email", "description": "User's email address (unique identifier)"}, {"name": "first_name", "required": True, "type": "string", "description": "First name"}, {"name": "last_name", "required": True, "type": "string", "description": "Last name"}, # Account settings {"name": "status", "required": False, "type": "enum", "description": "Account status", "options": ["active", "inactive", "pending_email", "pending_validation", "pre_validated", "payment_pending", "canceled", "expired"], "default": "active"}, {"name": "role", "required": False, "type": "enum", "description": "User role", "options": ["member", "admin", "finance", "guest"], "default": "member"}, # Contact info {"name": "phone", "required": False, "type": "string", "description": "Phone number"}, {"name": "address", "required": False, "type": "string", "description": "Street address"}, {"name": "city", "required": False, "type": "string", "description": "City"}, {"name": "state", "required": False, "type": "string", "description": "State/Province"}, {"name": "zipcode", "required": False, "type": "string", "description": "ZIP/Postal code"}, # Personal info {"name": "date_of_birth", "required": False, "type": "date", "description": "Date of birth (YYYY-MM-DD)"}, {"name": "member_since", "required": False, "type": "date", "description": "Member since date (YYYY-MM-DD)"}, # Partner info {"name": "partner_first_name", "required": False, "type": "string", "description": "Partner's first name"}, {"name": "partner_last_name", "required": False, "type": "string", "description": "Partner's last name"}, {"name": "partner_is_member", "required": False, "type": "boolean", "description": "Is partner a member? (true/false)"}, # Referral {"name": "referred_by_member_name", "required": False, "type": "string", "description": "Name of referring member"}, {"name": "lead_sources", "required": False, "type": "list", "description": "How they heard about us (comma-separated)"}, # Directory settings {"name": "show_in_directory", "required": False, "type": "boolean", "description": "Show in member directory? (true/false)", "default": "false"}, {"name": "directory_display_name", "required": False, "type": "string", "description": "Display name for directory"}, {"name": "directory_bio", "required": False, "type": "string", "description": "Bio for directory"}, {"name": "directory_email", "required": False, "type": "boolean", "description": "Show email in directory? (true/false)"}, {"name": "directory_phone", "required": False, "type": "boolean", "description": "Show phone in directory? (true/false)"}, {"name": "directory_address", "required": False, "type": "boolean", "description": "Show address in directory? (true/false)"}, # Newsletter preferences {"name": "newsletter_subscribed", "required": False, "type": "boolean", "description": "Subscribed to newsletter? (true/false)"}, {"name": "newsletter_publish_name", "required": False, "type": "boolean", "description": "Publish name in newsletter? (true/false)"}, {"name": "newsletter_publish_photo", "required": False, "type": "boolean", "description": "Publish photo in newsletter? (true/false)"}, {"name": "newsletter_publish_birthday", "required": False, "type": "boolean", "description": "Publish birthday in newsletter? (true/false)"}, # Volunteer interests {"name": "volunteer_interests", "required": False, "type": "list", "description": "Volunteer interests (comma-separated)"}, # Scholarship {"name": "scholarship_requested", "required": False, "type": "boolean", "description": "Requested scholarship? (true/false)"}, {"name": "scholarship_reason", "required": False, "type": "string", "description": "Reason for scholarship request"}, # Social media {"name": "social_media_facebook", "required": False, "type": "string", "description": "Facebook profile URL"}, {"name": "social_media_instagram", "required": False, "type": "string", "description": "Instagram handle"}, {"name": "social_media_twitter", "required": False, "type": "string", "description": "Twitter/X handle"}, {"name": "social_media_linkedin", "required": False, "type": "string", "description": "LinkedIn profile URL"}, ], "example_rows": [ { "email": "john.doe@example.com", "first_name": "John", "last_name": "Doe", "status": "active", "role": "member", "phone": "555-123-4567", "address": "123 Main St", "city": "Houston", "state": "TX", "zipcode": "77001", "date_of_birth": "1985-06-15", "member_since": "2023-01-01", "partner_first_name": "Jane", "partner_last_name": "Doe", "partner_is_member": "true", "referred_by_member_name": "", "lead_sources": "Friend,Social Media", "show_in_directory": "true", "directory_display_name": "John D.", "directory_bio": "Active community member", "directory_email": "true", "directory_phone": "false", "directory_address": "false", "newsletter_subscribed": "true", "newsletter_publish_name": "true", "newsletter_publish_photo": "false", "newsletter_publish_birthday": "true", "volunteer_interests": "Events,Outreach", "scholarship_requested": "false", "scholarship_reason": "", "social_media_facebook": "", "social_media_instagram": "@johndoe", "social_media_twitter": "", "social_media_linkedin": "", }, { "email": "jane.smith@example.com", "first_name": "Jane", "last_name": "Smith", "status": "active", "role": "member", "phone": "555-987-6543", "address": "456 Oak Ave", "city": "Dallas", "state": "TX", "zipcode": "75201", "date_of_birth": "1990-03-22", "member_since": "2024-06-15", "partner_first_name": "", "partner_last_name": "", "partner_is_member": "false", "referred_by_member_name": "John Doe", "lead_sources": "Current member", "show_in_directory": "false", "directory_display_name": "", "directory_bio": "", "directory_email": "false", "directory_phone": "false", "directory_address": "false", "newsletter_subscribed": "true", "newsletter_publish_name": "false", "newsletter_publish_photo": "false", "newsletter_publish_birthday": "false", "volunteer_interests": "", "scholarship_requested": "false", "scholarship_reason": "", "social_media_facebook": "", "social_media_instagram": "", "social_media_twitter": "", "social_media_linkedin": "", }, ] }, "subscriptions": { "name": "Subscriptions", "description": "Membership and subscription history for users", "required": False, "filename": "subscriptions_template.csv", "columns": [ {"name": "email", "required": True, "type": "email", "description": "User's email (must match users.csv)"}, {"name": "plan_name", "required": True, "type": "string", "description": "Name of the subscription plan"}, {"name": "status", "required": False, "type": "enum", "description": "Subscription status", "options": ["active", "cancelled", "expired", "past_due"], "default": "active"}, {"name": "amount", "required": False, "type": "decimal", "description": "Subscription amount paid"}, {"name": "currency", "required": False, "type": "string", "description": "Currency code (e.g., USD)", "default": "USD"}, {"name": "start_date", "required": False, "type": "date", "description": "Subscription start date (YYYY-MM-DD)"}, {"name": "end_date", "required": False, "type": "date", "description": "Subscription end date (YYYY-MM-DD)"}, {"name": "auto_renew", "required": False, "type": "boolean", "description": "Auto-renew enabled? (true/false)", "default": "false"}, {"name": "payment_method", "required": False, "type": "string", "description": "Payment method used"}, {"name": "notes", "required": False, "type": "string", "description": "Additional notes"}, # Stripe transaction fields {"name": "stripe_subscription_id", "required": False, "type": "string", "description": "Stripe Subscription ID (sub_xxx)"}, {"name": "stripe_customer_id", "required": False, "type": "string", "description": "Stripe Customer ID (cus_xxx)"}, {"name": "stripe_payment_intent_id", "required": False, "type": "string", "description": "Stripe Payment Intent ID (pi_xxx)"}, {"name": "stripe_invoice_id", "required": False, "type": "string", "description": "Stripe Invoice ID (in_xxx)"}, {"name": "stripe_charge_id", "required": False, "type": "string", "description": "Stripe Charge ID (ch_xxx)"}, {"name": "stripe_receipt_url", "required": False, "type": "string", "description": "Stripe receipt URL"}, {"name": "card_last4", "required": False, "type": "string", "description": "Last 4 digits of card used"}, {"name": "card_brand", "required": False, "type": "string", "description": "Card brand (Visa, Mastercard, etc.)"}, ], "example_rows": [ { "email": "john.doe@example.com", "plan_name": "Annual Membership", "status": "active", "amount": "100.00", "currency": "USD", "start_date": "2024-01-01", "end_date": "2024-12-31", "auto_renew": "true", "payment_method": "Credit Card", "notes": "", "stripe_subscription_id": "sub_1234567890", "stripe_customer_id": "cus_1234567890", "stripe_payment_intent_id": "pi_1234567890", "stripe_invoice_id": "in_1234567890", "stripe_charge_id": "ch_1234567890", "stripe_receipt_url": "https://pay.stripe.com/receipts/...", "card_last4": "4242", "card_brand": "Visa", }, { "email": "john.doe@example.com", "plan_name": "Annual Membership", "status": "expired", "amount": "85.00", "currency": "USD", "start_date": "2023-01-01", "end_date": "2023-12-31", "auto_renew": "false", "payment_method": "Check", "notes": "First year member discount", "stripe_subscription_id": "", "stripe_customer_id": "", "stripe_payment_intent_id": "", "stripe_invoice_id": "", "stripe_charge_id": "", "stripe_receipt_url": "", "card_last4": "", "card_brand": "", }, { "email": "jane.smith@example.com", "plan_name": "Monthly Membership", "status": "active", "amount": "15.00", "currency": "USD", "start_date": "2024-06-15", "end_date": "2024-07-15", "auto_renew": "true", "payment_method": "Credit Card", "notes": "", "stripe_subscription_id": "sub_0987654321", "stripe_customer_id": "cus_0987654321", "stripe_payment_intent_id": "pi_0987654321", "stripe_invoice_id": "", "stripe_charge_id": "ch_0987654321", "stripe_receipt_url": "", "card_last4": "1234", "card_brand": "Mastercard", }, ] }, "donations": { "name": "Donations", "description": "Donation history for users and anonymous donors", "required": False, "filename": "donations_template.csv", "columns": [ {"name": "email", "required": False, "type": "email", "description": "Donor's email (leave blank for anonymous)"}, {"name": "donor_name", "required": False, "type": "string", "description": "Donor's name (for anonymous donations)"}, {"name": "amount", "required": True, "type": "decimal", "description": "Donation amount"}, {"name": "currency", "required": False, "type": "string", "description": "Currency code", "default": "USD"}, {"name": "date", "required": True, "type": "date", "description": "Donation date (YYYY-MM-DD)"}, {"name": "type", "required": False, "type": "enum", "description": "Donation type", "options": ["member", "public"], "default": "member"}, {"name": "status", "required": False, "type": "enum", "description": "Donation status", "options": ["completed", "pending", "failed", "refunded"], "default": "completed"}, {"name": "payment_method", "required": False, "type": "string", "description": "Payment method"}, {"name": "campaign", "required": False, "type": "string", "description": "Campaign or fund name"}, {"name": "notes", "required": False, "type": "string", "description": "Additional notes"}, {"name": "is_recurring", "required": False, "type": "boolean", "description": "Is this a recurring donation?", "default": "false"}, # Stripe transaction fields {"name": "stripe_payment_intent_id", "required": False, "type": "string", "description": "Stripe Payment Intent ID (pi_xxx)"}, {"name": "stripe_charge_id", "required": False, "type": "string", "description": "Stripe Charge ID (ch_xxx)"}, {"name": "stripe_customer_id", "required": False, "type": "string", "description": "Stripe Customer ID (cus_xxx)"}, {"name": "stripe_checkout_session_id", "required": False, "type": "string", "description": "Stripe Checkout Session ID (cs_xxx)"}, {"name": "stripe_receipt_url", "required": False, "type": "string", "description": "Stripe receipt URL"}, {"name": "card_last4", "required": False, "type": "string", "description": "Last 4 digits of card used"}, {"name": "card_brand", "required": False, "type": "string", "description": "Card brand (Visa, Mastercard, etc.)"}, ], "example_rows": [ { "email": "john.doe@example.com", "donor_name": "", "amount": "50.00", "currency": "USD", "date": "2024-03-15", "type": "member", "status": "completed", "payment_method": "Credit Card", "campaign": "Annual Fund", "notes": "", "is_recurring": "false", "stripe_payment_intent_id": "pi_donation123", "stripe_charge_id": "ch_donation123", "stripe_customer_id": "cus_1234567890", "stripe_checkout_session_id": "", "stripe_receipt_url": "https://pay.stripe.com/receipts/...", "card_last4": "4242", "card_brand": "Visa", }, { "email": "", "donor_name": "Anonymous Supporter", "amount": "100.00", "currency": "USD", "date": "2024-04-01", "type": "public", "status": "completed", "payment_method": "Check", "campaign": "Building Fund", "notes": "Mailed check", "is_recurring": "false", "stripe_payment_intent_id": "", "stripe_charge_id": "", "stripe_customer_id": "", "stripe_checkout_session_id": "", "stripe_receipt_url": "", "card_last4": "", "card_brand": "", }, { "email": "jane.smith@example.com", "donor_name": "", "amount": "25.00", "currency": "USD", "date": "2024-05-01", "type": "member", "status": "completed", "payment_method": "Credit Card", "campaign": "", "notes": "Monthly recurring", "is_recurring": "true", "stripe_payment_intent_id": "pi_recurring456", "stripe_charge_id": "ch_recurring456", "stripe_customer_id": "cus_0987654321", "stripe_checkout_session_id": "cs_checkout789", "stripe_receipt_url": "", "card_last4": "1234", "card_brand": "Mastercard", }, ] }, "payments": { "name": "Payments", "description": "Payment transaction history", "required": False, "filename": "payments_template.csv", "columns": [ {"name": "email", "required": True, "type": "email", "description": "User's email (must match users.csv)"}, {"name": "amount", "required": True, "type": "decimal", "description": "Payment amount"}, {"name": "currency", "required": False, "type": "string", "description": "Currency code", "default": "USD"}, {"name": "date", "required": True, "type": "date", "description": "Payment date (YYYY-MM-DD)"}, {"name": "type", "required": False, "type": "enum", "description": "Payment type", "options": ["subscription", "donation", "event", "other"], "default": "subscription"}, {"name": "status", "required": False, "type": "enum", "description": "Payment status", "options": ["completed", "pending", "failed", "refunded"], "default": "completed"}, {"name": "payment_method", "required": False, "type": "string", "description": "Payment method (Credit Card, Check, Cash, etc.)"}, {"name": "description", "required": False, "type": "string", "description": "Payment description"}, {"name": "notes", "required": False, "type": "string", "description": "Internal notes"}, # Stripe transaction fields {"name": "stripe_payment_intent_id", "required": False, "type": "string", "description": "Stripe Payment Intent ID (pi_xxx)"}, {"name": "stripe_charge_id", "required": False, "type": "string", "description": "Stripe Charge ID (ch_xxx)"}, {"name": "stripe_customer_id", "required": False, "type": "string", "description": "Stripe Customer ID (cus_xxx)"}, {"name": "stripe_invoice_id", "required": False, "type": "string", "description": "Stripe Invoice ID (in_xxx)"}, {"name": "stripe_receipt_url", "required": False, "type": "string", "description": "Stripe receipt URL"}, {"name": "card_last4", "required": False, "type": "string", "description": "Last 4 digits of card used"}, {"name": "card_brand", "required": False, "type": "string", "description": "Card brand (Visa, Mastercard, etc.)"}, ], "example_rows": [ { "email": "john.doe@example.com", "amount": "100.00", "currency": "USD", "date": "2024-01-01", "type": "subscription", "status": "completed", "payment_method": "Credit Card", "description": "Annual Membership 2024", "notes": "", "stripe_payment_intent_id": "pi_sub123", "stripe_charge_id": "ch_sub123", "stripe_customer_id": "cus_1234567890", "stripe_invoice_id": "in_sub123", "stripe_receipt_url": "https://pay.stripe.com/receipts/...", "card_last4": "4242", "card_brand": "Visa", }, { "email": "john.doe@example.com", "amount": "50.00", "currency": "USD", "date": "2024-03-15", "type": "donation", "status": "completed", "payment_method": "Credit Card", "description": "Annual Fund Donation", "notes": "", "stripe_payment_intent_id": "pi_don456", "stripe_charge_id": "ch_don456", "stripe_customer_id": "cus_1234567890", "stripe_invoice_id": "", "stripe_receipt_url": "", "card_last4": "4242", "card_brand": "Visa", }, { "email": "jane.smith@example.com", "amount": "15.00", "currency": "USD", "date": "2024-06-15", "type": "subscription", "status": "completed", "payment_method": "Credit Card", "description": "Monthly Membership - June 2024", "notes": "", "stripe_payment_intent_id": "pi_mon789", "stripe_charge_id": "ch_mon789", "stripe_customer_id": "cus_0987654321", "stripe_invoice_id": "in_mon789", "stripe_receipt_url": "https://pay.stripe.com/receipts/...", "card_last4": "1234", "card_brand": "Mastercard", }, ] }, # Note: registration_data template is generated dynamically from the registration schema # See generate_dynamic_registration_template() function "registration_data": { "name": "Custom Registration Data", "description": "Custom registration fields based on current registration form configuration. This template is generated dynamically.", "required": False, "filename": "registration_data_template.csv", "is_dynamic": True, # Flag to indicate this template is generated dynamically "columns": [ {"name": "email", "required": True, "type": "email", "description": "User's email (must match users.csv)"}, # Additional columns are generated dynamically from registration schema ], "example_rows": [ { "email": "john.doe@example.com", # Example values are generated dynamically }, ] }, } # ============================================================================= # Template Generation Functions # ============================================================================= def generate_template_csv(template_type: str, include_examples: bool = True) -> str: """ Generate a CSV template string for the given template type. Args: template_type: One of 'users', 'subscriptions', 'donations', 'payments', 'registration_data' include_examples: Whether to include example data rows Returns: CSV string with headers and optionally example data """ if template_type not in TEMPLATES: raise ValueError(f"Unknown template type: {template_type}") template = TEMPLATES[template_type] columns = template["columns"] output = io.StringIO() writer = csv.writer(output) # Write header row headers = [col["name"] for col in columns] writer.writerow(headers) # Write example rows if requested if include_examples and template.get("example_rows"): for example in template["example_rows"]: row = [example.get(col["name"], "") for col in columns] writer.writerow(row) return output.getvalue() def generate_template_documentation(template_type: str) -> Dict[str, Any]: """ Generate documentation for a template including field descriptions. Args: template_type: One of the template types Returns: Dictionary with template metadata and field documentation """ if template_type not in TEMPLATES: raise ValueError(f"Unknown template type: {template_type}") template = TEMPLATES[template_type] return { "type": template_type, "name": template["name"], "description": template["description"], "required": template["required"], "filename": template["filename"], "fields": [ { "name": col["name"], "required": col["required"], "type": col["type"], "description": col["description"], "options": col.get("options"), "default": col.get("default"), } for col in template["columns"] ] } def get_all_templates_info() -> List[Dict[str, Any]]: """Get summary information about all available templates.""" return [ { "type": key, "name": template["name"], "description": template["description"], "required": template["required"], "filename": template["filename"], "field_count": len(template["columns"]), "required_fields": [col["name"] for col in template["columns"] if col["required"]], } for key, template in TEMPLATES.items() ] # ============================================================================= # CSV Parsing and Validation Functions # ============================================================================= def parse_boolean(value: Any) -> Optional[bool]: """Parse a boolean value from CSV.""" if value is None or value == "": return None if isinstance(value, bool): return value str_val = str(value).lower().strip() if str_val in ("true", "yes", "1", "y"): return True if str_val in ("false", "no", "0", "n"): return False return None def parse_date(value: Any) -> Optional[date]: """Parse a date value from CSV (expects YYYY-MM-DD).""" if value is None or value == "": return None try: if isinstance(value, date): return value return datetime.strptime(str(value).strip(), "%Y-%m-%d").date() except ValueError: # Try alternative formats for fmt in ["%m/%d/%Y", "%d/%m/%Y", "%Y/%m/%d"]: try: return datetime.strptime(str(value).strip(), fmt).date() except ValueError: continue return None def parse_decimal(value: Any) -> Optional[float]: """Parse a decimal value from CSV.""" if value is None or value == "": return None try: # Remove currency symbols and commas cleaned = re.sub(r'[,$€£]', '', str(value).strip()) return float(cleaned) except ValueError: return None def parse_list(value: Any) -> List[str]: """Parse a comma-separated list from CSV.""" if value is None or value == "": return [] return [item.strip() for item in str(value).split(",") if item.strip()] def validate_email(email: str) -> bool: """Validate email format.""" pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return bool(re.match(pattern, email.strip())) def validate_row(row: Dict[str, Any], template_type: str) -> Tuple[bool, List[str], Dict[str, Any]]: """ Validate a single row against the template schema. Args: row: Dictionary of column name -> value template_type: The template type to validate against Returns: Tuple of (is_valid, errors, parsed_row) """ if template_type not in TEMPLATES: return False, [f"Unknown template type: {template_type}"], {} template = TEMPLATES[template_type] columns = {col["name"]: col for col in template["columns"]} errors = [] parsed_row = {} # Check required fields for col_name, col_def in columns.items(): value = row.get(col_name, "") # Check required if col_def["required"] and (value is None or str(value).strip() == ""): errors.append(f"Required field '{col_name}' is missing") continue # Skip empty optional fields if value is None or str(value).strip() == "": parsed_row[col_name] = col_def.get("default") continue # Type validation and parsing col_type = col_def["type"] if col_type == "email": if not validate_email(str(value)): errors.append(f"Invalid email format in '{col_name}': {value}") else: parsed_row[col_name] = str(value).strip().lower() elif col_type == "string": parsed_row[col_name] = str(value).strip() elif col_type == "boolean": parsed = parse_boolean(value) if parsed is None and str(value).strip(): errors.append(f"Invalid boolean in '{col_name}': {value} (use true/false)") else: parsed_row[col_name] = parsed elif col_type == "date": parsed = parse_date(value) if parsed is None: errors.append(f"Invalid date in '{col_name}': {value} (use YYYY-MM-DD)") else: parsed_row[col_name] = parsed elif col_type == "decimal": parsed = parse_decimal(value) if parsed is None: errors.append(f"Invalid number in '{col_name}': {value}") else: parsed_row[col_name] = parsed elif col_type == "list": parsed_row[col_name] = parse_list(value) elif col_type == "enum": str_val = str(value).strip().lower() options = [opt.lower() for opt in col_def.get("options", [])] if str_val and str_val not in options: errors.append(f"Invalid value in '{col_name}': {value}. Must be one of: {', '.join(col_def['options'])}") else: parsed_row[col_name] = str_val if str_val else col_def.get("default") else: parsed_row[col_name] = value return len(errors) == 0, errors, parsed_row def parse_csv_file(file_content: bytes, template_type: str) -> Dict[str, Any]: """ Parse and validate a CSV file against a template. Args: file_content: Raw CSV file content template_type: The template type to validate against Returns: Dictionary with parsing results: { 'success': bool, 'total_rows': int, 'valid_rows': int, 'invalid_rows': int, 'rows': [...], # Parsed and validated rows 'errors': [...], # List of {row: int, errors: [...]} 'warnings': [...], } """ result = { 'success': True, 'total_rows': 0, 'valid_rows': 0, 'invalid_rows': 0, 'rows': [], 'errors': [], 'warnings': [], } try: # Decode file content try: content = file_content.decode('utf-8') except UnicodeDecodeError: content = file_content.decode('latin-1') # Parse CSV reader = csv.DictReader(io.StringIO(content)) # Check if headers match template template = TEMPLATES.get(template_type) if not template: result['success'] = False result['errors'].append({'row': 0, 'errors': [f"Unknown template type: {template_type}"]}) return result expected_headers = {col["name"] for col in template["columns"]} actual_headers = set(reader.fieldnames or []) # Check for missing required headers required_headers = {col["name"] for col in template["columns"] if col["required"]} missing_required = required_headers - actual_headers if missing_required: result['success'] = False result['errors'].append({ 'row': 0, 'errors': [f"Missing required columns: {', '.join(missing_required)}"] }) return result # Warn about unexpected headers unexpected = actual_headers - expected_headers if unexpected: result['warnings'].append(f"Unexpected columns will be ignored: {', '.join(unexpected)}") # Parse and validate each row for row_num, row in enumerate(reader, start=2): # Start at 2 (1 is header) result['total_rows'] += 1 is_valid, errors, parsed_row = validate_row(row, template_type) parsed_row['_row_number'] = row_num if is_valid: result['valid_rows'] += 1 result['rows'].append(parsed_row) else: result['invalid_rows'] += 1 result['errors'].append({ 'row': row_num, 'errors': errors, 'data': {k: v for k, v in row.items() if k in expected_headers} }) if result['invalid_rows'] > 0: result['success'] = False except Exception as e: logger.error(f"Error parsing CSV: {str(e)}") result['success'] = False result['errors'].append({'row': 0, 'errors': [f"Error parsing CSV: {str(e)}"]}) return result def cross_validate_files( users_result: Dict[str, Any], subscriptions_result: Optional[Dict[str, Any]] = None, donations_result: Optional[Dict[str, Any]] = None, payments_result: Optional[Dict[str, Any]] = None, registration_data_result: Optional[Dict[str, Any]] = None, existing_emails: Optional[set] = None ) -> Dict[str, Any]: """ Cross-validate multiple CSV files to ensure referential integrity. Checks: - All emails in related files exist in users.csv - No duplicate emails in users.csv - Emails don't already exist in database (if existing_emails provided) Returns: Validation result with warnings and errors """ result = { 'valid': True, 'warnings': [], 'errors': [], 'summary': { 'total_users': 0, 'new_users': 0, 'duplicate_users': 0, 'total_subscriptions': 0, 'total_donations': 0, 'total_payments': 0, 'total_registration_fields': 0, 'orphaned_records': 0, } } # Get all user emails from users.csv user_emails = set() duplicate_emails = set() for row in users_result.get('rows', []): email = row.get('email', '').lower() if email: if email in user_emails: duplicate_emails.add(email) user_emails.add(email) result['summary']['total_users'] = len(user_emails) result['summary']['duplicate_users'] = len(duplicate_emails) if duplicate_emails: result['errors'].append(f"Duplicate emails in users.csv: {', '.join(list(duplicate_emails)[:5])}{'...' if len(duplicate_emails) > 5 else ''}") result['valid'] = False # Check against existing database emails if existing_emails: existing_in_import = user_emails & existing_emails new_emails = user_emails - existing_emails result['summary']['new_users'] = len(new_emails) if existing_in_import: result['warnings'].append( f"{len(existing_in_import)} email(s) already exist in database and will be updated: " f"{', '.join(list(existing_in_import)[:3])}{'...' if len(existing_in_import) > 3 else ''}" ) # Validate subscriptions if subscriptions_result and subscriptions_result.get('rows'): result['summary']['total_subscriptions'] = len(subscriptions_result['rows']) orphaned = [] for row in subscriptions_result['rows']: email = row.get('email', '').lower() if email and email not in user_emails: orphaned.append(email) if orphaned: result['warnings'].append( f"{len(orphaned)} subscription(s) reference emails not in users.csv: " f"{', '.join(orphaned[:3])}{'...' if len(orphaned) > 3 else ''}" ) result['summary']['orphaned_records'] += len(orphaned) # Validate donations (email is optional for anonymous) if donations_result and donations_result.get('rows'): result['summary']['total_donations'] = len(donations_result['rows']) orphaned = [] for row in donations_result['rows']: email = row.get('email', '').lower() if email and email not in user_emails: orphaned.append(email) if orphaned: result['warnings'].append( f"{len(orphaned)} donation(s) reference emails not in users.csv: " f"{', '.join(orphaned[:3])}{'...' if len(orphaned) > 3 else ''}" ) result['summary']['orphaned_records'] += len(orphaned) # Validate payments if payments_result and payments_result.get('rows'): result['summary']['total_payments'] = len(payments_result['rows']) orphaned = [] for row in payments_result['rows']: email = row.get('email', '').lower() if email and email not in user_emails: orphaned.append(email) if orphaned: result['warnings'].append( f"{len(orphaned)} payment(s) reference emails not in users.csv: " f"{', '.join(orphaned[:3])}{'...' if len(orphaned) > 3 else ''}" ) result['summary']['orphaned_records'] += len(orphaned) # Validate registration data if registration_data_result and registration_data_result.get('rows'): result['summary']['total_registration_fields'] = len(registration_data_result['rows']) orphaned = [] for row in registration_data_result['rows']: email = row.get('email', '').lower() if email and email not in user_emails: orphaned.append(email) if orphaned: result['warnings'].append( f"{len(orphaned)} registration field(s) reference emails not in users.csv: " f"{', '.join(orphaned[:3])}{'...' if len(orphaned) > 3 else ''}" ) result['summary']['orphaned_records'] += len(orphaned) return result # ============================================================================= # Dynamic Registration Template Generator # ============================================================================= # Standard User model fields that have direct mappings (not stored in custom_registration_data) USER_MODEL_FIELDS = { 'email', 'first_name', 'last_name', 'phone', 'address', 'city', 'state', 'zipcode', 'date_of_birth', 'lead_sources', 'partner_first_name', 'partner_last_name', 'partner_is_member', 'partner_plan_to_become_member', 'referred_by_member_name', 'newsletter_subscribed', 'newsletter_publish_name', 'newsletter_publish_photo', 'newsletter_publish_birthday', 'newsletter_publish_none', 'volunteer_interests', 'scholarship_requested', 'scholarship_reason', 'show_in_directory', 'directory_email', 'directory_bio', 'directory_address', 'directory_phone', 'directory_dob', 'directory_partner_name', 'social_media_facebook', 'social_media_instagram', 'social_media_twitter', 'social_media_linkedin', 'accepts_tos', 'password' } def extract_custom_fields_from_schema(registration_schema: dict) -> List[Dict[str, Any]]: """ Extract custom fields from registration schema that should be stored in custom_registration_data. Custom fields are those where: - The field has no 'mapping' to a User model column, OR - The field's 'mapping' is not in USER_MODEL_FIELDS Args: registration_schema: The registration form schema from SystemSettings Returns: List of custom field definitions with their metadata """ custom_fields = [] for step in registration_schema.get('steps', []): for section in step.get('sections', []): for field in section.get('fields', []): field_id = field.get('id', '') mapping = field.get('mapping', field_id) # Skip fixed/core fields that map directly to User model if mapping in USER_MODEL_FIELDS: continue # Skip email/password/tos which are handled specially if field_id in ('email', 'password', 'confirm_password', 'accepts_tos'): continue # This is a custom field field_type = field.get('type', 'text') # Map registration field types to CSV column types csv_type = 'string' if field_type in ('number', 'currency'): csv_type = 'decimal' elif field_type == 'date': csv_type = 'date' elif field_type in ('checkbox', 'toggle'): csv_type = 'boolean' elif field_type in ('select', 'radio'): csv_type = 'enum' elif field_type in ('multiselect', 'checkbox_group'): csv_type = 'list' custom_field = { 'name': field_id, 'required': field.get('required', False), 'type': csv_type, 'label': field.get('label', field_id), 'description': field.get('placeholder', field.get('label', f'Custom field: {field_id}')), 'section': section.get('title', 'Unknown'), 'step': step.get('title', 'Unknown'), } # Add options for enum types if field_type in ('select', 'radio', 'multiselect', 'checkbox_group'): options = field.get('options', []) if options: custom_field['options'] = [opt.get('value', opt.get('label', '')) for opt in options] custom_fields.append(custom_field) return custom_fields def generate_dynamic_registration_template(registration_schema: dict) -> Dict[str, Any]: """ Generate a dynamic registration_data template based on the current registration schema. This creates a template where: - email column is always first and required - Each custom field becomes its own column (flat structure, not key-value pairs) - Column names match the field IDs from the registration schema Args: registration_schema: The registration form schema Returns: Template definition compatible with TEMPLATES format """ custom_fields = extract_custom_fields_from_schema(registration_schema) # Build columns list - email is always first columns = [ {"name": "email", "required": True, "type": "email", "description": "User's email (must match users.csv)"} ] # Add custom field columns for field in custom_fields: col = { "name": field['name'], "required": field['required'], "type": field['type'], "description": f"{field['label']} ({field['section']})", } if field.get('options'): col['options'] = field['options'] columns.append(col) # Generate example rows example_rows = [] if custom_fields: example_row = {"email": "john.doe@example.com"} for field in custom_fields: # Generate example value based on type if field['type'] == 'boolean': example_row[field['name']] = 'true' elif field['type'] == 'date': example_row[field['name']] = '2024-01-15' elif field['type'] == 'decimal': example_row[field['name']] = '100.00' elif field['type'] == 'list': opts = field.get('options', ['Option1', 'Option2']) example_row[field['name']] = ','.join(opts[:2]) if len(opts) > 1 else opts[0] if opts else '' elif field['type'] == 'enum': opts = field.get('options', []) example_row[field['name']] = opts[0] if opts else 'value' else: example_row[field['name']] = f'Example {field["label"]}' example_rows.append(example_row) # Add second example row example_row2 = {"email": "jane.smith@example.com"} for field in custom_fields: if field['type'] == 'boolean': example_row2[field['name']] = 'false' elif field['type'] == 'date': example_row2[field['name']] = '2023-06-20' elif field['type'] == 'decimal': example_row2[field['name']] = '50.00' elif field['type'] == 'list': opts = field.get('options', ['Option1']) example_row2[field['name']] = opts[0] if opts else '' elif field['type'] == 'enum': opts = field.get('options', []) example_row2[field['name']] = opts[1] if len(opts) > 1 else opts[0] if opts else '' else: example_row2[field['name']] = '' example_rows.append(example_row2) return { "name": "Custom Registration Data", "description": f"Custom registration fields from current form configuration ({len(custom_fields)} custom fields)", "required": False, "filename": "registration_data_template.csv", "is_dynamic": True, "columns": columns, "example_rows": example_rows, "custom_fields_info": [ { "name": f['name'], "label": f['label'], "type": f['type'], "required": f['required'], "section": f['section'], "step": f['step'], } for f in custom_fields ] } def generate_template_csv_dynamic(registration_schema: dict, include_examples: bool = True) -> str: """ Generate CSV content for the dynamic registration template. Args: registration_schema: The registration form schema include_examples: Whether to include example data rows Returns: CSV string content """ template = generate_dynamic_registration_template(registration_schema) columns = template["columns"] output = io.StringIO() writer = csv.writer(output) # Write header row headers = [col["name"] for col in columns] writer.writerow(headers) # Write example rows if requested if include_examples and template.get("example_rows"): for example in template["example_rows"]: row = [example.get(col["name"], "") for col in columns] writer.writerow(row) return output.getvalue()