TNCSC_Odoo/scripts/migrate_classroom.py

#!/usr/bin/env python3
"""Migrate a Google Classroom-style assignment list into a Community OS School
LMS channel (slide.channel), replacing Google Classroom.

Input CSV columns: title,url
  - title: the assignment/lesson title.
  - url:   a YouTube URL (embedded as a video slide) or any other link, e.g.
           a Google Drive document (created as an unpublished document slide
           with a note asking the admin to verify access / re-upload as PDF,
           since this script cannot read Google Drive content itself).

Uses Odoo's JSON-RPC API (not the Odoo MCP server) so it can run standalone
against any deployed instance. Re-running is safe: rows are matched by
(channel_id, name) and skipped if a slide with that title already exists in
the channel, unless --force is given.

Usage:
    python scripts/migrate_classroom.py \\
        --url https://your-odoo-host --db communityos_dev \\
        --username admin --password admin \\
        --channel-id 12 --csv data/raw/classroom_export.csv [--dry-run] [--force]
"""
import argparse
import csv
import json
import sys
import urllib.request
from pathlib import Path

YOUTUBE_HOSTS = ('youtube.com', 'youtu.be', 'www.youtube.com', 'm.youtube.com')


def jsonrpc_call(url, service, method, args):
    payload = {
        'jsonrpc': '2.0',
        'method': 'call',
        'params': {'service': service, 'method': method, 'args': args},
    }
    request = urllib.request.Request(
        f'{url}/jsonrpc',
        data=json.dumps(payload).encode('utf-8'),
        headers={'Content-Type': 'application/json'},
    )
    with urllib.request.urlopen(request) as response:
        result = json.loads(response.read())
    if 'error' in result:
        raise RuntimeError(result['error'].get('data', {}).get('message') or result['error'])
    return result['result']


def authenticate(url, db, username, password):
    uid = jsonrpc_call(url, 'common', 'authenticate', [db, username, password, {}])
    if not uid:
        raise RuntimeError('Authentication failed - check --url/--db/--username/--password')
    return uid


def execute_kw(url, db, uid, password, model, method, args, kwargs=None):
    full_args = [db, uid, password, model, method, args]
    if kwargs is not None:
        return jsonrpc_call(url, 'object', 'execute_kw', full_args + [kwargs])
    return jsonrpc_call(url, 'object', 'execute_kw', full_args)


def is_youtube(url_value):
    return any(host in url_value for host in YOUTUBE_HOSTS)


def read_rows(csv_path):
    with open(csv_path, newline='', encoding='utf-8') as handle:
        reader = csv.DictReader(handle)
        for row in reader:
            title = (row.get('title') or '').strip()
            link = (row.get('url') or '').strip()
            if title and link:
                yield title, link


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--url', required=True, help="Odoo base URL, e.g. http://localhost:8069")
    parser.add_argument('--db', required=True)
    parser.add_argument('--username', required=True)
    parser.add_argument('--password', required=True)
    parser.add_argument('--channel-id', required=True, type=int, help="Target slide.channel id")
    parser.add_argument('--csv', required=True, type=Path)
    parser.add_argument('--dry-run', action='store_true', help="Print what would be done, write nothing")
    parser.add_argument('--force', action='store_true', help="Recreate slides even if a same-titled one exists")
    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)

    existing_titles = set()
    if not args.force:
        existing = execute_kw(
            args.url, args.db, uid, args.password,
            'slide.slide', 'search_read',
            [[['channel_id', '=', args.channel_id]]], {'fields': ['name']},
        )
        existing_titles = {record['name'] for record in existing}

    created, skipped, errors = 0, 0, 0
    for title, link in read_rows(args.csv):
        if title in existing_titles:
            print(f"SKIP (already exists): {title}")
            skipped += 1
            continue

        if is_youtube(link):
            vals = {
                'name': title,
                'channel_id': args.channel_id,
                'slide_category': 'video',
                'source_type': 'external',
                'url': link,
                'is_published': True,
            }
            kind = 'video'
        else:
            vals = {
                'name': title,
                'channel_id': args.channel_id,
                'slide_category': 'document',
                'source_type': 'external',
                'url': link,
                'is_published': False,
                'description': (
                    "[MIGRATION] Imported from a non-YouTube link during Google Classroom "
                    "migration. Please verify access and re-upload as a PDF before publishing."
                ),
            }
            kind = 'document (needs review)'

        if args.dry_run:
            print(f"DRY-RUN would create {kind}: {title} -> {link}")
            created += 1
            continue

        try:
            execute_kw(args.url, args.db, uid, args.password, 'slide.slide', 'create', [vals])
            print(f"CREATED {kind}: {title}")
            created += 1
        except RuntimeError as exc:
            print(f"ERROR creating '{title}': {exc}", file=sys.stderr)
            errors += 1

    print(f"\nDone. created={created} skipped={skipped} errors={errors}")
    return 1 if errors else 0


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