Add daily Celery beat task and unversioned DOI tracking for preprint DOI resync

Author: Vlad0n20

api/crossref/views.py

@@ -49,9 +49,20 @@ def post(self, request):
                 if record.get('status').lower() == 'success' and doi:
                     msg = record.find('msg').text
                     created = bool(msg == 'Successfully added')
+                    # Unversioned DOIs (no _vN suffix, e.g. 10.31233/osf.io/tnaqp) are routing
+                    # aliases that always resolve to the latest version via OSF's GUID routing.
+                    # Store them as 'doi_unversioned' on the v1 preprint so we can track which
+                    # preprint series have had their unversioned DOI registered.
                     _, version = Guid.split_guid(guid) if guid else (None, None)
                     if not version:
-                        logger.info(f'Unversioned DOI confirmed by CrossRef (no identifier update needed): {doi}')
+                        logger.info(f'Unversioned DOI confirmed by CrossRef: {doi}')
+                        if created and guid:
+                            v1_preprint = Preprint.objects.filter(
+                                versioned_guids__guid___id=guid,
+                                versioned_guids__version=1,
+                            ).first()
+                            if v1_preprint:
+                                v1_preprint.set_identifier_value(category='doi_unversioned', value=doi)
                         dois_processed += 1
                         continue
 
@@ -74,9 +85,14 @@ def post(self, request):
                     if 'Relation target DOI does not exist' in record.find('msg').text:
                         logger.warning('Related publication DOI does not exist, sending metadata again without it...')
                         mint_doi_on_crossref_fail.apply_async(kwargs={'preprint_id': preprint._id})
-                    # This error occurs when a single preprint is being updated several times in a row with the same metadata [#PLAT-944]
-                    elif 'less or equal to previously submitted version' in record.find('msg').text and record_count == 2:
-                        break
+                    # This error occurs when a single preprint is being updated several times in a row
+                    # with the same metadata [#PLAT-944]. Previously this broke out of the loop when
+                    # record_count == 2 (single DOI submitted twice). Now batches legitimately contain
+                    # 2 records (versioned + unversioned DOI), so we continue instead of break to allow
+                    # the remaining record to be processed.
+                    elif 'less or equal to previously submitted version' in record.find('msg').text:
+                        dois_processed += 1
+                        continue
                     else:
                         unexpected_errors = True
             logger.info(f'Creation success email received from CrossRef for preprints: {guids}')

api_tests/crossref/views/test_crossref_email_response.py

@@ -263,3 +263,48 @@ def test_unversioned_doi_confirmation_skips_identifier_update(self, app, url, pr
 
         preprint.reload()
         assert preprint.get_identifier_value('doi') == versioned_doi
+        assert preprint.get_identifier_value('doi_unversioned') == unversioned_doi
+
+    def test_unversioned_doi_confirmation_update_does_not_store_doi_unversioned(self, app, url, preprint):
+        versioned_doi = settings.DOI_FORMAT.format(
+            prefix=preprint.provider.doi_prefix, guid=preprint._id
+        )
+        preprint.set_identifier_value(category='doi', value=versioned_doi)
+
+        base_guid = preprint.get_guid()._id
+        unversioned_doi = settings.DOI_FORMAT.format(
+            prefix=preprint.provider.doi_prefix, guid=base_guid
+        )
+        update_confirmation_xml = """
+            <?xml version="1.0" encoding="UTF-8"?>
+            <doi_batch_diagnostic status="completed" sp="cs3.crossref.org">
+               <submission_id>1390676000</submission_id>
+               <batch_id>{batch_id}</batch_id>
+               <record_diagnostic status="Success">
+                  <doi>{versioned_doi}</doi>
+                  <msg>Successfully updated</msg>
+               </record_diagnostic>
+               <record_diagnostic status="Success">
+                  <doi>{unversioned_doi}</doi>
+                  <msg>Successfully updated</msg>
+               </record_diagnostic>
+               <batch_data>
+                  <record_count>2</record_count>
+                  <success_count>2</success_count>
+                  <warning_count>0</warning_count>
+                  <failure_count>0</failure_count>
+               </batch_data>
+            </doi_batch_diagnostic>
+        """.format(
+            batch_id=preprint._id,
+            versioned_doi=versioned_doi,
+            unversioned_doi=unversioned_doi,
+        )
+
+        context_data = self.make_mailgun_payload(crossref_response=update_confirmation_xml)
+        with capture_notifications(expect_none=True):
+            app.post(url, context_data)
+
+        preprint.reload()
+        assert preprint.get_identifier_value('doi') == versioned_doi
+        assert preprint.get_identifier_value('doi_unversioned') is None

osf/management/commands/resync_preprint_dois_v1.py

@@ -5,6 +5,7 @@
 from django.core.management.base import BaseCommand
 from django.db.models import Q
 
+from framework.celery_tasks import app
 from osf.models import Preprint, Identifier
 from osf.models.base import VersionedGuidMixin
 from osf.management.commands.sync_doi_metadata import async_request_identifier_update
@@ -101,6 +102,114 @@ def resync_preprint_dois_v1(dry_run=True, batch_size=500, rate_limit=100, provid
         )
 
 
+def get_preprints_needing_unversioned_doi(provider_id=None):
+    content_type = ContentType.objects.get_for_model(Preprint)
+
+    already_has_unversioned = Identifier.objects.filter(
+        content_type=content_type,
+        category='doi_unversioned',
+        deleted__isnull=True,
+    ).values_list('object_id', flat=True)
+
+    has_versioned_doi = Identifier.objects.filter(
+        content_type=content_type,
+        category='doi',
+        deleted__isnull=True,
+        value__contains=VersionedGuidMixin.GUID_VERSION_DELIMITER,
+    ).values_list('object_id', flat=True)
+
+    public_query = Q(is_published=True, is_public=True, deleted__isnull=True)
+    withdrawn_query = Q(date_withdrawn__isnull=False, ever_public=True)
+
+    qs = Preprint.objects.filter(
+        versioned_guids__version=1,
+        id__in=has_versioned_doi,
+    ).filter(
+        public_query | withdrawn_query
+    ).exclude(
+        id__in=already_has_unversioned
+    ).exclude(
+        tags__name='qatest',
+        tags__system=True,
+    ).select_related('provider').distinct()
+
+    if provider_id:
+        qs = qs.filter(provider___id=provider_id)
+
+    return qs
+
+
+def register_missing_unversioned_dois(dry_run=True, batch_size=500, rate_limit=100, provider_id=None):
+    preprints_to_update = get_preprints_needing_unversioned_doi(provider_id=provider_id)
+
+    total = preprints_to_update.count()
+    logger.info(
+        f'{"[DRY RUN] " if dry_run else ""}'
+        f'{total} preprints need unversioned DOI registration'
+        + (f' (provider={provider_id})' if provider_id else '')
+    )
+
+    if batch_size:
+        preprints_iterable = preprints_to_update[:batch_size]
+    else:
+        preprints_iterable = preprints_to_update.iterator()
+
+    queued = 0
+    skipped = 0
+    errored = 0
+    for record_number, preprint in enumerate(preprints_iterable, 1):
+        if not preprint.provider.doi_prefix:
+            logger.warning(
+                f'Skipping preprint {preprint._id}: '
+                f'provider {preprint.provider._id} has no DOI prefix'
+            )
+            skipped += 1
+            continue
+
+        if dry_run:
+            logger.info(f'[DRY RUN] Would register unversioned DOI for preprint {preprint._id}')
+            queued += 1
+            continue
+
+        if rate_limit and not record_number % rate_limit:
+            logger.info(f'Rate limit reached at {record_number} preprints, sleeping {RATE_LIMIT_SLEEP}s')
+            time.sleep(RATE_LIMIT_SLEEP)
+
+        try:
+            async_request_identifier_update.apply_async(kwargs={'preprint_id': preprint._id})
+            logger.info(f'Queued unversioned DOI registration for preprint {preprint._id}')
+            queued += 1
+        except Exception:
+            logger.exception(f'Failed to queue unversioned DOI registration for preprint {preprint._id}')
+            errored += 1
+
+    logger.info(
+        f'{"[DRY RUN] " if dry_run else ""}'
+        f'Unversioned DOI pass done: {queued} queued, {skipped} skipped, {errored} errored'
+    )
+    if not dry_run and batch_size:
+        logger.info(
+            f'Estimated unversioned remaining after this batch: ~{max(0, total - queued - skipped - errored)}. '
+            f'Re-run until 0 preprints remain.'
+        )
+
+
+@app.task(name='osf.management.commands.resync_preprint_dois_v1', max_retries=0)
+def resync_preprint_dois_v1_task(batch_size=500, rate_limit=100, dry_run=False, provider_id=None):
+    resync_preprint_dois_v1(
+        dry_run=dry_run,
+        batch_size=batch_size,
+        rate_limit=rate_limit,
+        provider_id=provider_id,
+    )
+    register_missing_unversioned_dois(
+        dry_run=dry_run,
+        batch_size=batch_size,
+        rate_limit=rate_limit,
+        provider_id=provider_id,
+    )
+
+
 class Command(BaseCommand):
     help = (
         'Resync DOIs for version-1 preprints that are missing the versioned DOI suffix (_v1). '

website/identifiers/clients/crossref.py

@@ -148,7 +148,7 @@ def build_posted_content(self, preprint, element, include_relation, doi_override
         preprint_versions = preprint.get_preprint_versions(
             versioned_guids__version__lt=preprint.version,
             include_rejected=False,
-        )
+        ) if include_relation else []
         if preprint_versions:
             for previous_version in preprint_versions:
 

website/settings/defaults.py

@@ -711,6 +711,11 @@ class CeleryConfig:
             'schedule': crontab(minute=0, hour=5),  # Daily 12 a.m
             'kwargs': {'dry_run': False},
         },
+        'resync_preprint_dois_v1': {
+            'task': 'osf.management.commands.resync_preprint_dois_v1',
+            'schedule': crontab(minute=0, hour=5),  # Daily 12 a.m EDT
+            'kwargs': {'dry_run': False},
+        },
     }