Garante que o XML de um mesmo artigo receba o mesmo PID v3, independentemente da coleção de origem

article/models.py

@@ -2099,6 +2099,7 @@ def request_pid(self, user, force_update, auto_solve_pid_conflict):
             force_update=force_update,
             is_published=True,
             auto_solve_pid_conflict=auto_solve_pid_conflict,
+            collection_acron=self.collection and self.collection.acron3,
         )
 
         # Obtém a primeira resposta (assumindo apenas uma)

collection/migrations/0008_collection_network_classification_and_more.py

@@ -1,4 +1,4 @@
-# Generated by Django 5.2.7 on 2026-10-02 19:03
+# Generated by Django 5.2.7 on 2026-09-30 22:11
 
 import collection.models
 from django.db import migrations, models

collection/models.py

@@ -2,8 +2,11 @@
 from functools import cached_property
 
 from django import forms
+from django.conf import settings
 from django.contrib.postgres.fields import ArrayField
+from django.core.cache import cache
 from django.db import models
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from modelcluster.fields import ParentalKey
 from modelcluster.models import ClusterableModel
@@ -44,6 +47,26 @@ def formfield(self, **kwargs):
         return super(ArrayField, self).formfield(**defaults)
 
 
+ARTICLEMETA_COLLECTIONS_URL = (
+    "https://articlemeta.scielo.org/api/v1/collection/identifiers/"
+)
+
+ENSURE_NETWORK_CLASSIFICATION_CACHE_KEY = "collection.ensure_network_classification"
+# intervalo mínimo (segundos) entre consultas ao articlemeta feitas por
+# Collection.ensure_network_classification
+ENSURE_NETWORK_CLASSIFICATION_INTERVAL = 600
+
+
+def normalize_network_classification(network_classification):
+    """
+    Retorna network_classification como lista ou None
+    Ex.: "scielonetwork" -> ["scielonetwork"]
+    """
+    if isinstance(network_classification, str):
+        network_classification = [network_classification]
+    return [item for item in network_classification or [] if item] or None
+
+
 class CollectionName(TextWithLang):
     collection = ParentalKey(
         "Collection",
@@ -257,7 +280,7 @@ def __str__(self):
     def load(cls, user, collections_data=None, verify=False):
         if not collections_data:
             collections_data = fetch_data(
-                "https://articlemeta.scielo.org/api/v1/collection/identifiers/",
+                ARTICLEMETA_COLLECTIONS_URL,
                 json=True,
                 verify=verify,
             )
@@ -279,6 +302,88 @@ def load(cls, user, collections_data=None, verify=False):
                 network_classification=collection_data.get("network_classification"),
             )
 
+    @classmethod
+    def get_collections_without_network_classification(cls):
+        return cls.objects.filter(
+            models.Q(network_classification__isnull=True)
+            | models.Q(network_classification=[])
+        )
+
+    @classmethod
+    def ensure_network_classification(cls, user):
+        """
+        Garante network_classification das coleções antes de seu uso para
+        identificar a coleção principal (ex.: PidProviderXML.register de
+        artigos publicados em mais de uma coleção).
+
+        Consulta o articlemeta somente se há coleções sem esse dado e, no
+        máximo, uma vez a cada ENSURE_NETWORK_CLASSIFICATION_INTERVAL
+        segundos, evitando uma consulta por registro quando o articlemeta
+        está indisponível ou não tem a coleção.
+        Falhas são registradas no log e não interrompem quem chama.
+        """
+        if not getattr(settings, "COLLECTION_ENSURE_NETWORK_CLASSIFICATION", True):
+            return None
+        try:
+            if not cls.get_collections_without_network_classification().exists():
+                return None
+            # cache.add é atômico: somente um processo consulta o articlemeta
+            if not cache.add(
+                ENSURE_NETWORK_CLASSIFICATION_CACHE_KEY,
+                True,
+                ENSURE_NETWORK_CLASSIFICATION_INTERVAL,
+            ):
+                return None
+            return cls.complete_network_classification(user)
+        except Exception as e:
+            logging.exception(f"Collection.ensure_network_classification: {e}")
+            return None
+
+    @classmethod
+    def complete_network_classification(cls, user, collections_data=None, verify=False):
+        """
+        Preenche em lote network_classification das coleções que estão
+        sem esse dado, a partir dos dados do articlemeta.
+        Coleções já preenchidas não são alteradas.
+        """
+        queryset = cls.get_collections_without_network_classification()
+        result = {"updated": [], "not_found": []}
+        if not queryset.exists():
+            return result
+
+        if not collections_data:
+            collections_data = fetch_data(
+                ARTICLEMETA_COLLECTIONS_URL,
+                json=True,
+                verify=verify,
+            )
+        network_classification_by_acron = {
+            item.get("acron"): normalize_network_classification(
+                item.get("network_classification")
+            )
+            for item in collections_data
+        }
+
+        now = timezone.now()
+        items = []
+        for obj in queryset:
+            network_classification = network_classification_by_acron.get(obj.acron3)
+            if not network_classification:
+                result["not_found"].append(obj.acron3)
+                continue
+            obj.network_classification = network_classification
+            obj.updated_by = user
+            # bulk_update não aplica auto_now
+            obj.updated = now
+            items.append(obj)
+            result["updated"].append(obj.acron3)
+
+        cls.objects.bulk_update(
+            items, ["network_classification", "updated_by", "updated"]
+        )
+        logging.info(f"Collection.complete_network_classification: {result}")
+        return result
+
     @classmethod
     def get(cls, acron3):
         return cls.objects.get(acron3=acron3)
@@ -330,9 +435,9 @@ def create_or_update(
         obj.has_analytics = has_analytics
         obj.collection_type = collection_type
         obj.is_active = is_active
-        if isinstance(network_classification, str):
-            network_classification = [network_classification]
-        obj.network_classification = network_classification or None
+        obj.network_classification = normalize_network_classification(
+            network_classification
+        )
         obj.save()
         for language in names or {}:
             lang = Language.get_or_create(code2=language, creator=user)
@@ -369,6 +474,18 @@ def get_name_for_language(self, lang_code=None):
             self.collection_name.first().text if self.collection_name.exists() else ""
         )
 
+    @classmethod
+    def get_national_journal_collections(cls):
+        """
+        Retorna as coleções cuja classificação de rede
+        é exclusivamente scielonetwork
+        """
+        return cls.objects.filter(network_classification=["scielonetwork"])
+
+    @property
+    def is_national_journal_collection(self):
+        return self.network_classification == ["scielonetwork"]
+
     @classmethod
     def get_acronyms(cls, collection_acron_list):
         queryset = cls.objects

collection/scripts/complete_network_classification.py

@@ -0,0 +1,7 @@
+from collection import tasks
+
+
+def run(username):
+    tasks.task_complete_network_classification.apply_async(
+        kwargs={"username": username}
+    )

collection/tasks.py

@@ -7,12 +7,24 @@
 User = get_user_model()
 
 
+def _get_user(user_id=None, username=None):
+    if user_id:
+        return User.objects.get(pk=user_id)
+    if username:
+        return User.objects.get(username=username)
+    raise ValueError("user_id or username is required")
+
+
 @celery_app.task(bind=True)
 def task_load_collections(self, user_id=None, username=None):
-    if user_id:
-        user = User.objects.get(pk=user_id)
-    elif username:
-        user = User.objects.get(username=username)
-    else:
-        raise ValueError("user_id or username is required")
+    user = _get_user(user_id, username)
     Collection.load(user)
+
+
+@celery_app.task(bind=True)
+def task_complete_network_classification(self, user_id=None, username=None):
+    """
+    Preenche network_classification das coleções que estão sem esse dado
+    """
+    user = _get_user(user_id, username)
+    return Collection.complete_network_classification(user)

collection/tests/test_complete_network_classification.py

@@ -0,0 +1,85 @@
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+
+from collection.models import Collection
+
+User = get_user_model()
+
+
+class CompleteNetworkClassificationTest(TestCase):
+    def setUp(self):
+        self.user = User.objects.create(username="tester")
+        self.collections_data = [
+            {"acron": "scl", "network_classification": ["scielonetwork"]},
+            {"acron": "spa", "network_classification": ["scielonetwork", "thematic"]},
+            {"acron": "psi", "network_classification": "thematic"},
+            {"acron": "pef", "network_classification": ["independent"]},
+        ]
+
+    def test_completes_only_collections_without_network_classification(self):
+        Collection.objects.create(acron3="scl", creator=self.user)
+        Collection.objects.create(
+            acron3="spa", network_classification=[], creator=self.user
+        )
+        Collection.objects.create(acron3="psi", creator=self.user)
+        Collection.objects.create(
+            acron3="pef", network_classification=["thematic"], creator=self.user
+        )
+
+        result = Collection.complete_network_classification(
+            self.user, collections_data=self.collections_data
+        )
+
+        self.assertEqual(sorted(result["updated"]), ["psi", "scl", "spa"])
+        self.assertEqual(result["not_found"], [])
+        self.assertEqual(
+            Collection.objects.get(acron3="scl").network_classification,
+            ["scielonetwork"],
+        )
+        self.assertEqual(
+            Collection.objects.get(acron3="spa").network_classification,
+            ["scielonetwork", "thematic"],
+        )
+        self.assertEqual(
+            Collection.objects.get(acron3="psi").network_classification,
+            ["thematic"],
+        )
+        # já preenchida: mantém o valor atual
+        self.assertEqual(
+            Collection.objects.get(acron3="pef").network_classification,
+            ["thematic"],
+        )
+        self.assertEqual(Collection.objects.get(acron3="scl").updated_by, self.user)
+
+    def test_reports_collections_absent_from_source(self):
+        Collection.objects.create(acron3="xyz", creator=self.user)
+
+        result = Collection.complete_network_classification(
+            self.user, collections_data=self.collections_data
+        )
+
+        self.assertEqual(result, {"updated": [], "not_found": ["xyz"]})
+        self.assertIsNone(Collection.objects.get(acron3="xyz").network_classification)
+
+    @patch("collection.models.fetch_data")
+    def test_does_not_fetch_when_all_collections_are_complete(self, mock_fetch_data):
+        Collection.objects.create(
+            acron3="scl", network_classification=["scielonetwork"], creator=self.user
+        )
+
+        result = Collection.complete_network_classification(self.user)
+
+        self.assertEqual(result, {"updated": [], "not_found": []})
+        mock_fetch_data.assert_not_called()
+
+    @patch("collection.models.fetch_data")
+    def test_fetches_articlemeta_when_data_is_not_given(self, mock_fetch_data):
+        mock_fetch_data.return_value = self.collections_data
+        Collection.objects.create(acron3="scl", creator=self.user)
+
+        result = Collection.complete_network_classification(self.user)
+
+        self.assertEqual(result["updated"], ["scl"])
+        mock_fetch_data.assert_called_once()

collection/tests/test_ensure_network_classification.py

@@ -0,0 +1,73 @@
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.core.cache import cache
+from django.test import TestCase, override_settings
+
+from collection.models import ENSURE_NETWORK_CLASSIFICATION_CACHE_KEY, Collection
+
+User = get_user_model()
+
+
+@override_settings(COLLECTION_ENSURE_NETWORK_CLASSIFICATION=True)
+class EnsureNetworkClassificationTest(TestCase):
+    def setUp(self):
+        self.user = User.objects.create(username="tester")
+        cache.delete(ENSURE_NETWORK_CLASSIFICATION_CACHE_KEY)
+        self.addCleanup(cache.delete, ENSURE_NETWORK_CLASSIFICATION_CACHE_KEY)
+        patcher = patch(
+            "collection.models.fetch_data",
+            return_value=[
+                {"acron": "scl", "network_classification": ["scielonetwork"]},
+            ],
+        )
+        self.mock_fetch_data = patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def test_completes_collections_without_network_classification(self):
+        Collection.objects.create(acron3="scl", creator=self.user)
+
+        result = Collection.ensure_network_classification(self.user)
+
+        self.assertEqual(result, {"updated": ["scl"], "not_found": []})
+        self.assertTrue(
+            Collection.objects.get(acron3="scl").is_national_journal_collection
+        )
+        self.mock_fetch_data.assert_called_once()
+
+    def test_does_not_fetch_when_all_collections_are_complete(self):
+        Collection.objects.create(
+            acron3="scl", network_classification=["scielonetwork"], creator=self.user
+        )
+
+        self.assertIsNone(Collection.ensure_network_classification(self.user))
+        self.mock_fetch_data.assert_not_called()
+
+    def test_fetches_at_most_once_per_interval(self):
+        # coleção ausente no articlemeta continua sem network_classification
+        Collection.objects.create(acron3="xyz", creator=self.user)
+
+        result = Collection.ensure_network_classification(self.user)
+        self.assertEqual(result, {"updated": [], "not_found": ["xyz"]})
+        self.assertIsNone(Collection.ensure_network_classification(self.user))
+        self.mock_fetch_data.assert_called_once()
+
+        # após o intervalo, consulta novamente
+        cache.delete(ENSURE_NETWORK_CLASSIFICATION_CACHE_KEY)
+        Collection.ensure_network_classification(self.user)
+        self.assertEqual(self.mock_fetch_data.call_count, 2)
+
+    def test_does_not_raise_when_articlemeta_fails(self):
+        self.mock_fetch_data.side_effect = Exception("articlemeta indisponível")
+        Collection.objects.create(acron3="scl", creator=self.user)
+
+        self.assertIsNone(Collection.ensure_network_classification(self.user))
+        self.assertIsNone(Collection.objects.get(acron3="scl").network_classification)
+
+    @override_settings(COLLECTION_ENSURE_NETWORK_CLASSIFICATION=False)
+    def test_does_nothing_when_disabled(self):
+        Collection.objects.create(acron3="scl", creator=self.user)
+
+        self.assertIsNone(Collection.ensure_network_classification(self.user))
+        self.mock_fetch_data.assert_not_called()
+        self.assertIsNone(Collection.objects.get(acron3="scl").network_classification)