TNCSC_Odoo/scripts/migrate_students.py

#!/usr/bin/env python3
"""Migrate existing TNCSC school students (+ their parent/guardian) into
Community OS School.

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):

    student_full_name,student_date_of_birth,level_code,proficiency,grade_ref,
    parent_full_name,parent_email,parent_phone,emergency_contact_name,
    emergency_contact_phone,health_notes

  - student_full_name: required.
  - parent_full_name / parent_email: required. parent_email is the dedupe/
    match key for the parent (case-insensitive) - if a partner with that
    email already exists (e.g. created by migrate_members.py), it is reused
    rather than duplicated.
  - student_date_of_birth: optional, several common date formats accepted
    and normalized to ISO (YYYY-MM-DD).
  - level_code: optional, must match an existing community.school.level
    code. Used to place the student into that level's class in the CURRENT
    term (the community.school.term with state='open', most recent
    start_date if more than one). If blank, unmatched, or no such class
    exists in the current term, the student record is still created/updated
    but no enrollment is made (reported as 'no_enrollment' with a reason).
  - proficiency: optional, one of beginner/intermediate/advanced (defaults
    to beginner if blank or unrecognized).
  - parent_phone / emergency_contact_phone: optional, normalized to digits
    (with a leading + kept if present).

Behaviour:
  - A student has no email of their own in most source exports, so there is
    no natural unique key for them the way email works for members. Instead
    a student is matched by (parent_partner_id, student name) - re-running
    the same file updates the existing student rather than creating a
    duplicate.
  - Rows missing student_full_name/parent_full_name/parent_email, or that
    duplicate a (parent_email, student_full_name) pair already seen earlier
    in the SAME file, are marked 'failed' and skipped - they never reach
    Odoo.
  - Enrollment is only created once per (student, class) pair; re-running
    does not create a second enrollment.

Usage:
    python scripts/migrate_students.py \\
        --url https://your-odoo-host --db communityos_dev \\
        --username admin --password admin \\
        --csv data/raw/students_export.csv \\
        --report data/raw/students_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,
)

PROFICIENCIES = ('beginner', 'intermediate', 'advanced')


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


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)

    current_term = rpc(
        'community.school.term', 'search_read',
        [[('state', '=', 'open')]], {'fields': ['id'], 'order': 'start_date desc', 'limit': 1},
    )
    current_term_id = current_term[0]['id'] if current_term else None

    level_by_code = {}
    for level in rpc('community.school.level', 'search_read', [[]], {'fields': ['code']}):
        level_by_code[level['code']] = level['id']

    seen = set()

    for row_number, row in enumerate(read_rows(args.csv), start=2):
        student_name = (row.get('student_full_name') or '').strip()
        parent_name = (row.get('parent_full_name') or '').strip()
        parent_email = normalize_email(row.get('parent_email'))
        key = f"{parent_email or 'row ' + str(row_number)} / {student_name or '?'}"

        if not student_name or not parent_name or not parent_email:
            report.add(key, 'failed', 'missing student_full_name, parent_full_name, or parent_email')
            continue
        dedupe_key = (parent_email, student_name.lower())
        if dedupe_key in seen:
            report.add(key, 'failed', 'duplicate (parent_email, student_full_name) in source file')
            continue
        seen.add(dedupe_key)

        proficiency = (row.get('proficiency') or '').strip().lower()
        if proficiency not in PROFICIENCIES:
            proficiency = 'beginner'

        level_code = (row.get('level_code') or '').strip()
        level_id = level_by_code.get(level_code) if level_code else None

        if args.dry_run:
            action = 'would_create_or_update'
            reason = ''
            if level_code and not level_id:
                reason = f"unknown level_code '{level_code}' (student still created/updated)"
            elif not current_term_id:
                reason = 'no open term - no enrollment' if level_code else ''
            report.add(key, action, reason)
            continue

        parent_ids = rpc('res.partner', 'search', [[('email', '=', parent_email)]], {'limit': 1})
        if parent_ids:
            parent_id = parent_ids[0]
            rpc('res.partner', 'write', [[parent_id], {
                'name': parent_name,
                'phone': normalize_phone(row.get('parent_phone')),
            }])
        else:
            parent_id = rpc('res.partner', 'create', [{
                'name': parent_name,
                'email': parent_email,
                'phone': normalize_phone(row.get('parent_phone')),
            }])

        existing_student = rpc(
            'community.school.student', 'search_read',
            [[('parent_partner_id', '=', parent_id)]], {'fields': ['id', 'partner_id']},
        )
        existing_student = [s for s in existing_student if s['partner_id'] and s['partner_id'][1] == student_name]

        student_vals = {
            'proficiency': proficiency,
            'grade_ref': (row.get('grade_ref') or '').strip() or False,
            'health_notes': (row.get('health_notes') or '').strip() or False,
            'emergency_contact_name': (row.get('emergency_contact_name') or '').strip() or False,
            'emergency_contact_phone': normalize_phone(row.get('emergency_contact_phone')),
        }
        dob = normalize_date(row.get('student_date_of_birth'))
        if dob:
            student_vals['date_of_birth'] = dob

        if existing_student:
            student_id = existing_student[0]['id']
            student_partner_id = existing_student[0]['partner_id'][0]
            rpc('res.partner', 'write', [[student_partner_id], {'name': student_name}])
            rpc('community.school.student', 'write', [[student_id], student_vals])
            action = 'updated'
        else:
            student_partner_id = rpc('res.partner', 'create', [{'name': student_name}])
            student_vals.update({'partner_id': student_partner_id, 'parent_partner_id': parent_id})
            student_id = rpc('community.school.student', 'create', [student_vals])
            action = 'created'

        reason = ''
        if level_code and not level_id:
            reason = f"unknown level_code '{level_code}' - no enrollment"
        elif level_id and not current_term_id:
            reason = 'no open term - no enrollment'
        elif level_id and current_term_id:
            class_ids = rpc('community.school.class', 'search', [[
                ('level_id', '=', level_id), ('term_id', '=', current_term_id),
            ]], {'limit': 1})
            if not class_ids:
                reason = f"no class for level '{level_code}' in current term - no enrollment"
            else:
                class_id = class_ids[0]
                enrollment_ids = rpc('community.school.enrollment', 'search', [[
                    ('student_id', '=', student_id), ('class_id', '=', class_id),
                ]], {'limit': 1})
                if not enrollment_ids:
                    rpc('community.school.enrollment', 'create', [{
                        'student_id': student_id, 'class_id': class_id,
                    }])
                    action += '+enrolled'

        report.add(key, action, reason)

    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())