Garante o registro do documento na coleção de origem do XML e guarda o nome do pacote SPS por coleção

article/tests/test_article_by_collection.py

@@ -0,0 +1,222 @@
+"""
+Um Article por CollectionPidV2 (XML + coleção, mesmo pid v3): filtros por
+coleção, relatórios por coleção, um documento por Article no índice de
+busca e vínculo entre as instâncias do mesmo artigo no OAI.
+"""
+from importlib import import_module
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+
+from article.models import Article
+from article.search_indexes import ArticleIndex, ArticleOAIIndex
+from article.tests.test_mixins import make_article_in_collection
+from collection.models import Collection
+from journal.models import Journal, SciELOJournal
+from report import data_availability
+
+User = get_user_model()
+
+PID_V3 = "CVcmNHzKpFmYM77VQBJwHYt"
+PID_V2_SCL = "S0103-65642009000300003"
+PID_V2_PSI = "S1678-51772009000300003"
+
+
+class ArticleByCollectionDataMixin:
+    """
+    Psicologia USP: scl (pusp, 0103-6564) e psi (psicousp, 1678-5177)
+    Journal Single: somente scl (single, 0000-0001)
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.user = User.objects.create_user(username="by-collection", password="x")
+        cls.scl = Collection.objects.create(
+            acron3="scl",
+            network_classification=["scielonetwork"],
+            domain="www.scielo.br",
+            is_active=True,
+            creator=cls.user,
+        )
+        cls.psi = Collection.objects.create(
+            acron3="psi",
+            domain="pepsic.bvsalud.org",
+            is_active=True,
+            creator=cls.user,
+        )
+        cls.pusp = Journal.objects.create(title="Psicologia USP", creator=cls.user)
+        cls.single = Journal.objects.create(title="Journal Single", creator=cls.user)
+        for journal, collection, acron, issn_scielo in (
+            (cls.pusp, cls.scl, "pusp", "0103-6564"),
+            (cls.pusp, cls.psi, "psicousp", "1678-5177"),
+            (cls.single, cls.scl, "single", "0000-0001"),
+        ):
+            SciELOJournal.objects.create(
+                journal=journal,
+                collection=collection,
+                journal_acron=acron,
+                issn_scielo=issn_scielo,
+                creator=cls.user,
+            )
+        cls.scl_item = make_article_in_collection(
+            cls.user, PID_V3, cls.scl, PID_V2_SCL, "pusp",
+            journal=cls.pusp, is_classic_public=True,
+        )
+        cls.psi_item = make_article_in_collection(
+            cls.user, PID_V3, cls.psi, PID_V2_PSI, "psicousp",
+            journal=cls.pusp, is_classic_public=True,
+        )
+
+
+class SelectItemsByCollectionTest(ArticleByCollectionDataMixin, TestCase):
+    def test_collection_acron_list(self):
+        self.assertEqual(
+            set(Article.select_items(collection_acron_list=["psi"])),
+            {self.psi_item},
+        )
+        self.assertEqual(
+            set(Article.select_items(collection_acron_list=["scl", "psi"])),
+            {self.scl_item, self.psi_item},
+        )
+
+    def test_journal_acron_of_the_collection(self):
+        self.assertEqual(
+            set(Article.select_items(journal_acron_list=["psicousp"])),
+            {self.psi_item},
+        )
+        self.assertEqual(
+            set(Article.select_items(journal_acron_list=["pusp"])),
+            {self.scl_item},
+        )
+
+    def test_journal_acron_and_collection(self):
+        self.assertEqual(
+            list(Article.select_items(
+                collection_acron_list=["scl"], journal_acron_list=["psicousp"]
+            )),
+            [],
+        )
+
+    def test_data_availability_report_counts_by_collection(self):
+        for collection, expected in ((self.scl, 1), (self.psi, 1), (None, 2)):
+            qs = data_availability.filter_articles(collection=collection)
+            self.assertEqual(qs.count(), expected, collection)
+
+
+class ArticleIndexByCollectionTest(ArticleByCollectionDataMixin, TestCase):
+    """Um documento por Article (coleção), relacionados pelo pid v3."""
+
+    def test_index_queryset_has_one_item_by_collection(self):
+        self.assertEqual(
+            set(ArticleIndex().index_queryset()), {self.scl_item, self.psi_item}
+        )
+
+    def test_collection_fields_of_the_article_collection(self):
+        index = ArticleIndex()
+        self.assertEqual(index.prepare_collection(self.psi_item), ["psi"])
+        self.assertEqual(index.prepare_ids(self.psi_item)[:2], [PID_V2_PSI, PID_V3])
+        self.assertEqual(index.prepare_ta_cluster(self.psi_item), ["psicousp"])
+
+    def test_oai_identifier_has_collection(self):
+        index = ArticleOAIIndex()
+        self.assertEqual(
+            index.prepare_id(self.psi_item), f"oai:scielo:psi_{PID_V2_PSI}"
+        )
+        self.assertEqual(
+            index.prepare_item_id(self.scl_item), f"oai:scielo:scl_{PID_V2_SCL}"
+        )
+        self.assertEqual(index.prepare_collections(self.psi_item), {"1678-5177"})
+
+    def test_identical_items_urls(self):
+        self.assertEqual(
+            self.scl_item.identical_items_urls,
+            [f"{self.psi.base_url}/scielo.php?script=sci_arttext&pid={PID_V2_PSI}"],
+        )
+
+    def test_oai_compile_relates_identical_items(self):
+        compiled = ArticleOAIIndex().fields["compile"].prepare(self.scl_item)
+        self.assertIn('<element name="isidenticalto">', compiled)
+        self.assertIn(
+            f"{self.psi.base_url}/scielo.php?script=sci_arttext&amp;pid={PID_V2_PSI}",
+            compiled,
+        )
+        self.assertIn(
+            f'<field name="identifier">scl_{PID_V2_SCL}</field>', compiled
+        )
+        self.assertIn(f'<field name="identifier">{PID_V3}</field>', compiled)
+        # o identificador local vem primeiro
+        self.assertLess(
+            compiled.index(f"scl_{PID_V2_SCL}</field>"),
+            compiled.index(f'<field name="identifier">{PID_V3}</field>'),
+        )
+        # identificadores pid v2 e pid v3 qualificados
+        self.assertRegex(
+            compiled,
+            r'<element name="pidv2">\s*<element>\s*<field name="value">'
+            + PID_V2_SCL
+            + "</field>",
+        )
+        self.assertRegex(
+            compiled,
+            r'<element name="pidv3">\s*<element>\s*<field name="value">'
+            + PID_V3
+            + "</field>",
+        )
+
+
+class MigrationIdentifyCollectionTest(TestCase):
+    """article.0054: identificação da coleção dos artigos"""
+
+    def setUp(self):
+        self.migration = import_module(
+            "article.migrations.0054_migrate_article_collection_pid_v2"
+        )
+        main = ["scielonetwork"]
+        self.journal_collections = {
+            # periódico em uma coleção
+            1: [{"collection_id": 10, "issn_scielo": "0000-0001",
+                 "journal_acron": "single", "collection__network_classification": main}],
+            # issn_scielo diferente em cada coleção
+            2: [
+                {"collection_id": 10, "issn_scielo": "0103-6564",
+                 "journal_acron": "pusp", "collection__network_classification": main},
+                {"collection_id": 20, "issn_scielo": "1678-5177",
+                 "journal_acron": "psicousp", "collection__network_classification": None},
+            ],
+            # mesmo issn_scielo nas duas coleções
+            3: [
+                {"collection_id": 20, "issn_scielo": "0102-311X",
+                 "journal_acron": "csp", "collection__network_classification": None},
+                {"collection_id": 10, "issn_scielo": "0102-311X",
+                 "journal_acron": "csp", "collection__network_classification": main},
+            ],
+            # mesmo issn_scielo, nenhuma coleção principal
+            4: [
+                {"collection_id": 20, "issn_scielo": "0254-9247",
+                 "journal_acron": "rp", "collection__network_classification": None},
+                {"collection_id": 30, "issn_scielo": "0254-9247",
+                 "journal_acron": "rp", "collection__network_classification": None},
+            ],
+        }
+
+    def identify(self, journal_id, pid_v2=None, collection_id=None):
+        return self.migration.identify_collection_id(
+            {"journal_id": journal_id, "pid_v2": pid_v2, "collection_id": collection_id},
+            self.journal_collections,
+        )
+
+    def test_article_collection(self):
+        self.assertEqual(self.identify(2, PID_V2_SCL, collection_id=20), 20)
+
+    def test_single_collection_journal(self):
+        self.assertEqual(self.identify(1), 10)
+
+    def test_issn_scielo_of_pid_v2(self):
+        self.assertEqual(self.identify(2, PID_V2_SCL), 10)
+        self.assertEqual(self.identify(2, PID_V2_PSI), 20)
+
+    def test_same_issn_scielo_uses_main_collection(self):
+        self.assertEqual(self.identify(3, "S0102-311X2019000104001"), 10)
+
+    def test_not_identified(self):
+        self.assertIsNone(self.identify(4, "S0254-92472019000104001"))

article/tests/test_migration_fill_article_collection.py

@@ -0,0 +1,126 @@
+from importlib import import_module
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.db import connection
+from django.test import TestCase
+
+from article.models import AMArticle, Article
+from collection.models import Collection
+from journal.models import Journal, SciELOJournal
+
+User = get_user_model()
+
+migration = import_module("article.migrations.0055_fill_article_collection")
+
+PID_V2_SCL = "S0103-65642009000300003"
+PID_V2_PSI = "S1678-51772009000300003"
+
+
+class FillArticleCollectionMigrationTest(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username="fill_collection_user")
+        self.scl = Collection.objects.create(
+            acron3="scl", network_classification=["scielonetwork"], creator=self.user
+        )
+        self.psi = Collection.objects.create(acron3="psi", creator=self.user)
+        self.rve = Collection.objects.create(acron3="rve", creator=self.user)
+
+        # Psicologia USP: issn_scielo diferente em cada coleção
+        self.pusp = self.make_journal(
+            "pusp", ((self.scl, "0103-6564"), (self.psi, "1678-5177"))
+        )
+        # somente em scl
+        self.single = self.make_journal("single", ((self.scl, "0000-0001"),))
+        # mesmo issn_scielo em scl e psi
+        self.csp = self.make_journal(
+            "csp", ((self.psi, "0102-311X"), (self.scl, "0102-311X"))
+        )
+        # mesmo issn_scielo, nenhuma coleção principal
+        self.rp = self.make_journal(
+            "rp", ((self.psi, "0254-9247"), (self.rve, "0254-9247"))
+        )
+
+    def make_journal(self, title, collections):
+        journal = Journal.objects.create(title=title, creator=self.user)
+        for collection, issn_scielo in collections:
+            SciELOJournal.objects.create(
+                journal=journal,
+                collection=collection,
+                issn_scielo=issn_scielo,
+                creator=self.user,
+            )
+        return journal
+
+    def make_article(self, pid_v3, journal, pid_v2=None, collection=None):
+        return Article.objects.create(
+            pid_v3=pid_v3,
+            pid_v2=pid_v2,
+            journal=journal,
+            collection=collection,
+            creator=self.user,
+        )
+
+    def run_migration(self, batch_size=2):
+        with patch.object(migration, "BATCH_SIZE", batch_size), patch("builtins.print"):
+            migration.fill_article_collection(
+                None, SimpleNamespace(connection=connection)
+            )
+
+    def assert_collection(self, article, expected):
+        article.refresh_from_db()
+        self.assertEqual(article.collection, expected, article.pid_v3)
+
+    def test_legacy_article_with_same_pid_v2(self):
+        article = self.make_article("v3-legacy", self.pusp, PID_V2_PSI)
+        AMArticle.objects.create(
+            pid=PID_V2_SCL, collection=self.scl, new_record=article, creator=self.user
+        )
+        AMArticle.objects.create(
+            pid=PID_V2_PSI, collection=self.psi, new_record=article, creator=self.user
+        )
+        self.run_migration()
+        self.assert_collection(article, self.psi)
+
+    def test_issn_scielo_of_pid_v2(self):
+        scl_item = self.make_article("v3-scl", self.pusp, PID_V2_SCL)
+        psi_item = self.make_article("v3-psi", self.pusp, PID_V2_PSI)
+        self.run_migration()
+        self.assert_collection(scl_item, self.scl)
+        self.assert_collection(psi_item, self.psi)
+
+    def test_same_issn_scielo_uses_main_collection(self):
+        article = self.make_article("v3-csp", self.csp, "S0102-311X2019000104001")
+        self.run_migration()
+        self.assert_collection(article, self.scl)
+
+    def test_single_collection_journal(self):
+        article = self.make_article("v3-single", self.single)
+        self.run_migration()
+        self.assert_collection(article, self.scl)
+
+    def test_main_collection_of_the_journal(self):
+        article = self.make_article("v3-main", self.pusp)
+        self.run_migration()
+        self.assert_collection(article, self.scl)
+
+    def test_not_identified(self):
+        article = self.make_article("v3-rp", self.rp, "S0254-92472019000104001")
+        no_journal = self.make_article("v3-no-journal", None)
+        self.run_migration()
+        self.assert_collection(article, None)
+        self.assert_collection(no_journal, None)
+
+    def test_keeps_existing_collection(self):
+        article = self.make_article("v3-kept", self.pusp, PID_V2_SCL, self.psi)
+        self.run_migration()
+        self.assert_collection(article, self.psi)
+
+    def test_many_batches(self):
+        articles = [
+            self.make_article(f"v3-{i}", self.pusp, PID_V2_PSI) for i in range(7)
+        ]
+        self.run_migration(batch_size=3)
+        for article in articles:
+            self.assert_collection(article, self.psi)

pid_provider/base_pid_provider.py

@@ -408,7 +408,7 @@ def is_registered_xml_zip(cls, zip_xml_file_path):
                     ),
                 },
             )
-            return {
+            yield {
                 "error_msg": f"Unable to check whether {zip_xml_file_path} is registered {e}",
                 "error_type": str(type(e)),
             }

pid_provider/exceptions.py

@@ -48,4 +48,9 @@ class APIPidProviderFixPidV2Error(Exception): ...
 class UnmatchedPidProviderXMLError(Exception): ...
 
 
-class RequiredCollectionErrorToGetPidProviderXMLError(Exception): ...
+class RequiredCollectionErrorToGetPidProviderXMLError(Exception): ...
+
+
+class XMLCollectionNotFoundErrorToGetPidProviderXMLError(
+    RequiredCollectionErrorToGetPidProviderXMLError
+): ...

pid_provider/migrations/0021_collectionpidv2_sps_pkg_name.py

@@ -0,0 +1,26 @@
+# Generated by Django 5.2.7 on 2026-10-28 13:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("pid_provider", "0020_collectionpidv2"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="collectionpidv2",
+            name="sps_pkg_name",
+            field=models.CharField(
+                blank=True, max_length=100, null=True, verbose_name="SPS package name"
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="collectionpidv2",
+            index=models.Index(
+                fields=["sps_pkg_name"], name="pid_provide_sps_pkg_8e5c4f_idx"
+            ),
+        ),
+    ]