videos: submitter migration

Author: zubeydecivelek

cds_migrator_kit/rdm/users/api.py

@@ -10,48 +10,24 @@
 import json
 from copy import deepcopy
 
-from flask import current_app
+from cds_migrator_kit.users.api import MigrationUserAPI
+
+from invenio_db import db
 from invenio_accounts.models import User, UserIdentity
 from invenio_cern_sync.sso import cern_remote_app_name
-from invenio_db import db
-from invenio_oauthclient.models import RemoteAccount
-from invenio_rdm_migrator.transform.base import Entry, Transform
-from invenio_userprofiles import UserProfile
 from psycopg.errors import UniqueViolation
 from sqlalchemy.exc import IntegrityError
 
 from cds_migrator_kit.transform.dumper import CDSRecordDump
 
 
-class CDSMigrationUserAPI:
+
+class CDSMigrationUserAPI(MigrationUserAPI):
     """CDS missing user load class."""
 
     def __init__(self, remote_account_client_id=None):
         """Constructor."""
-        self.client_id = current_app.config["CERN_APP_CREDENTIALS"]["consumer_key"]
-
-    def check_person_id_exists(self, person_id):
-        """Check if uer identity already exists."""
-        return UserIdentity.query.filter_by(id=person_id).one_or_none()
-
-    def create_invenio_user(self, email, username):
-        """Commit new user in db."""
-        try:
-            user = User(email=email, username=username, active=False)
-            db.session.add(user)
-            db.session.commit()
-            return user
-        except IntegrityError as e:
-            db.session.rollback()
-            email_username = email.split("@")[0]
-            user = User(
-                email=email,
-                username=f"duplicated_{username}_{email_username}",
-                active=False,
-            )
-            db.session.add(user)
-            db.session.commit()
-            return user
+        super().__init__(remote_account_client_id)
 
     def create_invenio_user_identity(self, user_id, person_id):
         """Return new user identity entry."""
@@ -72,40 +48,3 @@ def create_invenio_user_identity(self, user_id, person_id):
             db.session.commit()
             return user_identity
 
-    def create_invenio_user_profile(self, user, name):
-        """Return new user profile."""
-        user_profile = UserProfile(user=user)
-        user_profile.full_name = name
-        return user_profile
-
-    def create_invenio_remote_account(self, user_id, extra_data=None):
-        """Return new user entry."""
-        if extra_data is None:
-            extra_data = {}
-        return RemoteAccount.create(
-            client_id=self.client_id, user_id=user_id, extra_data=extra_data
-        )
-
-    def create_user(self, email, name, person_id, username, extra_data=None):
-        """Create an invenio user."""
-        user = self.create_invenio_user(email, username)
-        user_id = user.id
-        profile_data = {}
-        if person_id:
-            identity = self.create_invenio_user_identity(user_id, person_id)
-            db.session.add(identity)
-            profile_data = {
-                "person_id": person_id,
-            }
-        if name:
-            if "department" in extra_data:
-                profile_data.update({"department": extra_data["department"]})
-            profile = deepcopy(user.user_profile)
-            profile.update(profile_data)
-            user.user_profile = profile
-            db.session.add(user)
-
-        remote_account = self.create_invenio_remote_account(user_id, extra_data)
-        db.session.add(remote_account)
-
-        return user

cds_migrator_kit/users/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2025 CERN.
+#
+# cds-migrator-kit is free software; you can redistribute it and/or modify it under
+# the terms of the MIT License; see LICENSE file for more details.
+
+"""cds-migrator-kit users migration module."""

cds_migrator_kit/users/api.py

@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2024-2025 CERN.
+#
+# Ccds-migrator-kit is free software; you can redistribute it and/or modify it under
+# the terms of the MIT License; see LICENSE file for more details.
+
+"""cds-migrator-kit user api."""
+import csv
+import json
+from copy import deepcopy
+
+from flask import current_app
+from invenio_accounts.models import User, UserIdentity
+from invenio_db import db
+from invenio_oauthclient.models import RemoteAccount
+from invenio_userprofiles import UserProfile
+from psycopg.errors import UniqueViolation
+from sqlalchemy.exc import IntegrityError
+
+from cds_migrator_kit.transform.dumper import CDSRecordDump
+
+cern_remote_app_name = "cern" 
+
+class MigrationUserAPI:
+    """CDS missing user load class."""
+
+    def __init__(self, remote_account_client_id=None):
+        """Constructor."""
+        self.client_id = current_app.config["CERN_APP_CREDENTIALS"]["consumer_key"]
+
+    def check_person_id_exists(self, person_id):
+        """Check if uer identity already exists."""
+        return UserIdentity.query.filter_by(id=person_id).one_or_none()
+
+    def create_invenio_user(self, email, username):
+        """Commit new user in db."""
+        try:
+            user = User(email=email, username=username, active=False)
+            db.session.add(user)
+            db.session.commit()
+            return user
+        except IntegrityError as e:
+            db.session.rollback()
+            email_username = email.split("@")[0]
+            user = User(
+                email=email,
+                username=f"duplicated_{username}_{email_username}",
+                active=False,
+            )
+            db.session.add(user)
+            db.session.commit()
+            return user
+
+    def create_invenio_user_identity(self, user_id, person_id):
+        """Return new user identity entry."""
+        try:
+            return UserIdentity(
+                id=person_id,
+                method=cern_remote_app_name,
+                id_user=user_id,
+            )
+        except (IntegrityError, UniqueViolation) as e:
+            db.session.rollback()
+            user_identity = UserIdentity(
+                id=f"duplicate{person_id}",
+                method=cern_remote_app_name,
+                id_user=user_id,
+            )
+            db.session.add(user_identity)
+            db.session.commit()
+            return user_identity
+
+    def create_invenio_user_profile(self, user, name):
+        """Return new user profile."""
+        user_profile = UserProfile(user=user)
+        user_profile.full_name = name
+        return user_profile
+
+    def create_invenio_remote_account(self, user_id, extra_data=None):
+        """Return new user entry."""
+        if extra_data is None:
+            extra_data = {}
+        return RemoteAccount.create(
+            client_id=self.client_id, user_id=user_id, extra_data=extra_data
+        )
+
+    def create_user(self, email, name, person_id, username, extra_data=None):
+        """Create an invenio user."""
+        user = self.create_invenio_user(email, username)
+        user_id = user.id
+        profile_data = {}
+        if person_id:
+            identity = self.create_invenio_user_identity(user_id, person_id)
+            db.session.add(identity)
+            profile_data = {
+                "person_id": person_id,
+            }
+        if name:
+            if "department" in extra_data:
+                profile_data.update({"department": extra_data["department"]})
+            profile = deepcopy(user.user_profile)
+            profile.update(profile_data)
+            user.user_profile = profile
+            db.session.add(user)
+
+        remote_account = self.create_invenio_remote_account(user_id, extra_data)
+        db.session.add(remote_account)
+
+        return user

cds_migrator_kit/videos/README.md

@@ -10,7 +10,7 @@ Run the following command to dump a subset of records:
 
     inveniomigrator dump records -q '8567_x:"Absolute master path" 8567_d:/mnt/master_share* -980__C:MIGRATED -980__c:DELETED -5831_a:digitized' --file-prefix lectures --chunk-size=1000
 
-Place your dumps into the `cds_migrator_kit/videos/weblecture_migration/data/weblectures/dump/` folder, or update the `extract: / dirpath:` in `cds_migrator_kit/videos/weblecture_migration/streams.yaml`.
+Place your dumps into the `cds_migrator_kit/videos/weblecture_migration/data/weblectures/dump/` folder, or update the `records/weblectures/extract/dirpath:` in `cds_migrator_kit/videos/weblecture_migration/streams.yaml`.
 
 For the files, modify the `migration_config.py` file located at `cds_migrator_kit/videos/migration_config.py`, specifically updating the following variables:
 
@@ -22,6 +22,26 @@ For the files, modify the `migration_config.py` file located at `cds_migrator_ki
 - Composite videos will always be named as `<id-composite-...p-quality.mp4>`, and frames of the composite will be stored in `MOUNTED_MEDIA_CEPH_PATH/frames`.
 - If no composite exists (i.e., the master contains only one video), subformats and frames will be obtained using `data.v2.json`.
 
+## Missing Users Migration
+
+If you need to migrate missing users, you need `missing_users.json` file, you can find the file in the folder:` eos\...\CDS Videos\Projects\Weblectures migration\`. Place the file in `cds_migrator_kit/videos/weblecture_migration/data/users/` or update the `submitters/data_dir` in `cds_migrator_kit/videos/weblecture_migration/streams.yaml`. User migration is also using the same record dumps with record migration, so make sure you have your dumps in `cds_migrator_kit/videos/weblecture_migration/data/weblectures/dump/` folder.
+
+To migrate the missing users, run:
+
+.. code-block:: bash
+
+    invenio migration videos submitters run
+
+## Creating System User
+
+If you need to use a system user for the migration, you can configure the username by updating the `WEBLECTURES_MIGRATION_SYSTEM_USER` variable in the `migration_config.py` file.
+
+To create the system user (if it doesn't already exist), run:
+
+.. code-block:: bash
+
+    invenio migration videos submitters create-system-user
+
 ## Running the Migration
 
 Once you have the dump and files, you can proceed with the migration.

cds_migrator_kit/videos/migration_config.py

@@ -50,3 +50,5 @@ def _(x):  # needed to avoid start time failure with lazy strings
 
 # TODO CHANGE THEM
 MOUNTED_MEDIA_CEPH_PATH = "/cephfs/media_data"
+
+WEBLECTURES_MIGRATION_SYSTEM_USER = "[email protected]"

cds_migrator_kit/videos/weblecture_migration/cli.py

@@ -8,14 +8,24 @@
 """CDS-Videos command line module."""
 import logging
 from pathlib import Path
+import re
 
+from cds_migrator_kit.videos.weblecture_migration.users.api import CDSVideosMigrationUserAPI
 import click
 from flask import current_app
 from flask.cli import with_appcontext
+from invenio_accounts.models import User
+from sqlalchemy.exc import NoResultFound
 
 from cds_migrator_kit.runner.runner import Runner
-from cds_migrator_kit.videos.weblecture_migration.streams import RecordStreamDefinition
 from cds_migrator_kit.videos.weblecture_migration.logger import VideosJsonLogger
+from cds_migrator_kit.videos.weblecture_migration.streams import RecordStreamDefinition
+from cds_migrator_kit.videos.weblecture_migration.users.runner import (
+    VideosSubmitterRunner,
+)
+from cds_migrator_kit.videos.weblecture_migration.users.streams import (
+    SubmitterStreamDefinition,
+)
 
 cli_logger = logging.getLogger("migrator")
 
@@ -40,13 +50,69 @@ def weblectures():
 @with_appcontext
 def run(dry_run=False):
     """Run."""
-    # TODO temporary, it'll change
-    VideosJsonLogger.initialize(Path("cds_migrator_kit/videos/weblecture_migration/log"))
     stream_config = current_app.config["CDS_MIGRATOR_KIT_VIDEOS_STREAM_CONFIG"]
     runner = Runner(
         stream_definitions=[RecordStreamDefinition],
         config_filepath=Path(stream_config).absolute(),
         dry_run=dry_run,
         collection="weblectures",
     )
+    VideosJsonLogger.initialize(runner.log_dir)
+    runner.run()
+
+
+@videos.group()
+def submitters():
+    """Migration CLI for weblectures."""
+    pass
+
+
+@submitters.command()
+@click.option(
+    "--dry-run",
+    is_flag=True,
+)
+@with_appcontext
+def run(dry_run=False):
+    """Migrate the users(submitters) if missing."""
+    stream_config = current_app.config["CDS_MIGRATOR_KIT_VIDEOS_STREAM_CONFIG"]
+    runner = VideosSubmitterRunner(
+        stream_definition=SubmitterStreamDefinition,
+        config_filepath=Path(stream_config).absolute(),
+        dry_run=dry_run,
+    )
     runner.run()
+    
+
+@submitters.command()
+@with_appcontext
+def create_system_user():
+    """Create the sytem user if missing."""
+    email = current_app.config["WEBLECTURES_MIGRATION_SYSTEM_USER"] 
+    
+    if not email:
+        cli_logger.error("System user email is not configured.")
+        return
+    
+    try:
+        user = User.query.filter_by(email=email).one()
+        cli_logger.info(f"User {email} exists.")
+        return
+    except NoResultFound:
+        username = email.split("@")[0].replace(".", "")
+        username = re.sub(r"\W+", "", username).lower()
+        extra_data = {"migration": {"note": "System user for migration"}}
+        user_api = CDSVideosMigrationUserAPI()
+        
+        try:
+            user = user_api.create_user(
+                email,
+                name="Weblectures System User",
+                username=username,
+                person_id=None,
+                extra_data=extra_data,
+            )
+        except Exception as exc:
+            cli_logger.error(
+                f"System user creation failed: {email}, {username}\n {exc}"
+            )