TNCSC_Odoo/scripts/migrate_members.py

#!/usr/bin/env python3
"""Migrate existing TNCSC members into Community OS Membership.

Input CSV columns (source system export - adjust the header names below to
match the real export once available, the column *names* are what matter,
not their order):

    full_name,email,phone,tier_code,membership_start,membership_expiry,is_volunteer

  - full_name: required.
  - email: required, used as the dedupe/match key.
  - phone: optional, normalized to digits (with a leading + kept if present).
  - tier_code: required, must match an existing community.membership.tier
    code (e.g. IND, FAM, STU, SEN, LIFE for the TNCSC deployment).
  - membership_start / membership_expiry: optional, several common date
    formats are accepted and normalized to ISO (YYYY-MM-DD).
  - is_volunteer: optional, "1"/"true"/"yes" (case-insensitive) = True.

Behaviour:
  - Rows missing full_name/email, with an unrecognized tier_code, or that
    duplicate an email already seen earlier in the SAME file are marked
    'failed' with a reason and skipped - they never reach Odoo.
  - Existing members are matched by email (case-insensitive) and updated;
    new ones are created. Re-running the same file is safe: it always
    converges on the same end state rather than creating duplicates.
  - A member with a tier and no explicit state that isn't already active
    is activated (generates a member ID) after being created/updated.

Usage:
    python scripts/migrate_members.py \\
        --url https://your-odoo-host --db communityos_dev \\
        --username admin --password admin \\
        --csv data/raw/members_export.csv \\
        --report data/raw/members_migration_report.csv \\
        [--dry-run]
"""
import argparse
import csv
import sys
from pathlib import Path

from _migration_common import (
    MigrationReport, authenticate, execute_kw, normalize_date, normalize_email, normalize_phone,
)


def read_rows(csv_path):
    with open(csv_path, newline='', encoding='utf-8') as handle:
        yield from csv.DictReader(handle)


def is_truthy(value):
    return (value or '').strip().lower() in ('1', 'true', 'yes', 'y')


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--url', required=True)
    parser.add_argument('--db', required=True)
    parser.add_argument('--username', required=True)
    parser.add_argument('--password', required=True)
    parser.add_argument('--csv', required=True, type=Path)
    parser.add_argument('--report', required=True, type=Path)
    parser.add_argument('--dry-run', action='store_true')
    args = parser.parse_args()

    if not args.csv.exists():
        print(f"CSV file not found: {args.csv}", file=sys.stderr)
        return 1

    uid = authenticate(args.url, args.db, args.username, args.password)
    report = MigrationReport()

    def rpc(model, method, call_args, kwargs=None):
        return execute_kw(args.url, args.db, uid, args.password, model, method, call_args, kwargs)

    tier_by_code = {}
    for tier in rpc('community.membership.tier', 'search_read', [[]], {'fields': ['code']}):
        tier_by_code[tier['code']] = tier['id']

    seen_emails = set()

    for row_number, row in enumerate(read_rows(args.csv), start=2):
        name = (row.get('full_name') or '').strip()
        email = normalize_email(row.get('email'))
        key = email or f'row {row_number}'

        if not name or not email:
            report.add(key, 'failed', 'missing full_name or email')
            continue
        if email in seen_emails:
            report.add(key, 'failed', 'duplicate email in source file')
            continue
        seen_emails.add(email)

        tier_code = (row.get('tier_code') or '').strip()
        tier_id = tier_by_code.get(tier_code)
        if tier_code and not tier_id:
            report.add(key, 'failed', f"unknown tier_code '{tier_code}'")
            continue

        vals = {
            'name': name,
            'email': email,
            'phone': normalize_phone(row.get('phone')),
            'is_volunteer': is_truthy(row.get('is_volunteer')),
        }
        if tier_id:
            vals['membership_tier_id'] = tier_id
        start = normalize_date(row.get('membership_start'))
        if start:
            vals['membership_start'] = start
        expiry = normalize_date(row.get('membership_expiry'))
        if expiry:
            vals['membership_expiry'] = expiry

        existing = rpc('res.partner', 'search', [[('email', '=', email)]], {'limit': 1})

        if args.dry_run:
            report.add(key, 'would_update' if existing else 'would_create')
            continue

        if existing:
            rpc('res.partner', 'write', [existing, vals])
            partner_id = existing[0]
            action = 'updated'
        else:
            partner_id = rpc('res.partner', 'create', [vals])
            action = 'created'

        if tier_id:
            partner_state = rpc('res.partner', 'read', [[partner_id], ['membership_state']])[0]['membership_state']
            if partner_state not in ('active', 'renewal_due'):
                rpc('res.partner', 'action_activate_membership', [[partner_id]])

        report.add(key, action)

    report.write(args.report)
    summary = report.summary()
    print(f"Done. {summary}")
    print(f"Report written to {args.report}")
    return 1 if summary.get('failed') else 0


if __name__ == '__main__':
    sys.exit(main())