#!/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())