Processamento de coleta dos dados de periódico e fascículo na API do core.

config/settings/base.py

@@ -323,8 +323,16 @@
     "CORE_JOURNAL_API_ENDPOINT",
     default="/api/v2/pid/journal/",
 )
+CORE_ISSUE_API_ENDPOINT = env(
+    "CORE_ISSUE_API_ENDPOINT",
+    default="/api/v1/issue/",
+)
+CORE_ISSUE_FROM_DATE_CREATED = env(
+    "CORE_ISSUE_FROM_DATE_CREATED",
+    default="2019-01-01",
+)
 CORE_COLLECTION_API_URL = f"{CORE_API_DOMAIN}{CORE_COLLECTION_API_ENDPOINT}"
 CORE_JOURNAL_API_URL = f"{CORE_API_DOMAIN}{CORE_JOURNAL_API_ENDPOINT}"
 
 #Aumento en el límite de campos
-DATA_UPLOAD_MAX_NUMBER_FIELDS = 10000
+DATA_UPLOAD_MAX_NUMBER_FIELDS = 10000

config/urls.py

@@ -1,17 +1,16 @@
+from core.search import views as search_views
 from django.conf import settings
-from django.urls import include, path
-from django.contrib import admin
-from django.conf.urls.static import static
 from django.conf.urls.i18n import i18n_patterns  # ← Adicionar esta linha
-from wagtail.admin import urls as wagtailadmin_urls
+from django.conf.urls.static import static
+from django.contrib import admin
+from django.urls import include, path
+from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
 from wagtail import urls as wagtail_urls
+from wagtail.admin import urls as wagtailadmin_urls
 from wagtail.documents import urls as wagtaildocs_urls
-from wagtailautocomplete.urls.admin import urlpatterns as autocomplete_admin_urls
-from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
 
-from core.search import views as search_views
-from reference import views as reference_views
 from config import api_router as api_router
+from markup_doc.autocomplete import urlpatterns as autocomplete_admin_urls
 
 urlpatterns = [
     path("django-admin/", admin.site.urls),
@@ -20,10 +19,12 @@
     path("search/", search_views.search, name="search"),
     # JWT
     path("api/v1/auth/token/", TokenObtainPairView.as_view(), name="token_obtain_pair"),
-    path("api/v1/auth/token/refresh/", TokenRefreshView.as_view(), name="token_refresh"),
+    path(
+        "api/v1/auth/token/refresh/", TokenRefreshView.as_view(), name="token_refresh"
+    ),
     path("api/v1/", include(api_router)),
     # URL para trocar idioma
-    path('i18n/', include('django.conf.urls.i18n')),
+    path("i18n/", include("django.conf.urls.i18n")),
 ] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)
 
 # URLs com prefixo de idioma

core/migrations/0003_coresyncstate_alter_flexibledate_id_alter_gender_id_and_more.py

@@ -0,0 +1,51 @@
+# Generated by Django 6.0.5 on 2026-05-31 17:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0002_wagtailsearch_indexentry_text_defaults'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CoreSyncState',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('resource', models.CharField(max_length=50, unique=True, verbose_name='Resource')),
+                ('last_updated_at', models.DateTimeField(blank=True, null=True, verbose_name='Last updated at')),
+                ('last_success_at', models.DateTimeField(blank=True, null=True, verbose_name='Last success at')),
+            ],
+            options={
+                'verbose_name': 'Core sync state',
+                'verbose_name_plural': 'Core sync states',
+            },
+        ),
+        migrations.AlterField(
+            model_name='flexibledate',
+            name='id',
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+        ),
+        migrations.AlterField(
+            model_name='gender',
+            name='id',
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+        ),
+        migrations.AlterField(
+            model_name='language',
+            name='id',
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+        ),
+        migrations.AlterField(
+            model_name='license',
+            name='id',
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+        ),
+        migrations.AlterField(
+            model_name='licensestatement',
+            name='id',
+            field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+        ),
+    ]

core/models.py

@@ -1,18 +1,20 @@
 import os
-from django.db import models, IntegrityError
-from django.db.models import Case, When, Value, IntegerField
+
 from django.contrib.auth import get_user_model
+from django.db import IntegrityError, models
+from django.db.models import Case, IntegerField, Value, When
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from wagtail.admin.panels import FieldPanel
 from wagtail.fields import RichTextField
-from wagtail.search import index
 from wagtailautocomplete.edit_handlers import AutocompletePanel
 
 from . import choices
 from .utils.utils import language_iso
 
 User = get_user_model()
 
+
 class CommonControlField(models.Model):
     """
     Class with common control fields.
@@ -77,11 +79,9 @@ def autocomplete_label(self):
         FieldPanel("gender"),
     ]
 
-
     class Meta:
         unique_together = [("code", "gender")]
 
-
     def __unicode__(self):
         return self.gender or self.code
 
@@ -224,17 +224,20 @@ def get_object_in_preferred_language(self, language):
         mission = self.filter(language=language)
         if mission:
             return mission
-        
-        language_order = ['pt', 'es', 'en']
+
+        language_order = ["pt", "es", "en"]
         langs = self.all().values_list("language", flat=True)
         languages = Language.objects.filter(id__in=langs)
-        
+
         # Define a ordem baseado na lista language_order
-        order = [When(code2=lang, then=Value(i)) for i, lang in enumerate(language_order)]
+        order = [
+            When(code2=lang, then=Value(i)) for i, lang in enumerate(language_order)
+        ]
         ordered_languages = languages.annotate(
-            language_order=Case(*order, default=Value(len(language_order)), output_field=IntegerField())
-        ).order_by('language_order')
-
+            language_order=Case(
+                *order, default=Value(len(language_order)), output_field=IntegerField()
+            )
+        ).order_by("language_order")
 
         for lang in ordered_languages:
             mission = self.filter(language=lang)
@@ -257,7 +260,7 @@ class RichTextWithLanguage(models.Model):
         AutocompletePanel("language"),
         FieldPanel("rich_text"),
     ]
-    
+
     objects = LanguageFallbackManager()
 
     class Meta:
@@ -297,7 +300,7 @@ def autocomplete_label(self):
     ]
 
     class Meta:
-        unique_together = [("license_type", )]
+        unique_together = [("license_type",)]
         verbose_name = _("License")
         verbose_name_plural = _("Licenses")
         indexes = [
@@ -326,9 +329,7 @@ def get(
     ):
         if not license_type:
             raise ValueError("License.get requires license_type parameters")
-        filters = dict(
-            license_type__iexact=license_type
-        )
+        filters = dict(license_type__iexact=license_type)
         try:
             return cls.objects.get(**filters)
         except cls.MultipleObjectsReturned:
@@ -368,7 +369,8 @@ class LicenseStatement(CommonControlField):
         Language, on_delete=models.SET_NULL, null=True, blank=True
     )
     license = models.ForeignKey(
-        License, on_delete=models.SET_NULL, null=True, blank=True)
+        License, on_delete=models.SET_NULL, null=True, blank=True
+    )
 
     panels = [
         FieldPanel("url"),
@@ -406,7 +408,8 @@ def get(
             raise ValueError("LicenseStatement.get requires url or license_p")
         try:
             return cls.objects.get(
-                url__iexact=url, license_p__iexact=license_p, language=language)
+                url__iexact=url, license_p__iexact=license_p, language=language
+            )
         except cls.MultipleObjectsReturned:
             return cls.objects.filter(
                 url__iexact=url, license_p__iexact=license_p, language=language
@@ -447,9 +450,7 @@ def create_or_update(
     ):
         try:
             data = dict(
-                url=url,
-                license_p=license_p,
-                language=language and language.code2
+                url=url, license_p=license_p, language=language and language.code2
             )
             try:
                 obj = cls.get(url, license_p, language)
@@ -464,7 +465,9 @@ def create_or_update(
             except cls.DoesNotExist:
                 return cls.create(user, url, license_p, language, license)
         except Exception as e:
-            raise ValueError(f"Unable to create or update LicenseStatement for {data}: {type(e)} {e}")
+            raise ValueError(
+                f"Unable to create or update LicenseStatement for {data}: {type(e)} {e}"
+            )
 
     @staticmethod
     def parse_url(url):
@@ -513,7 +516,7 @@ class FileWithLang(models.Model):
         blank=True,
         on_delete=models.SET_NULL,
         verbose_name=_("File"),
-        help_text='',
+        help_text="",
         related_name="+",
     )
 
@@ -536,3 +539,52 @@ def filename(self):
 
     class Meta:
         abstract = True
+
+
+class CoreSyncState(models.Model):
+    """
+    Guarda o checkpoint da última coleta da API Core por recurso.
+
+    A próxima coleta deve sempre retomar a partir de ``last_updated_at``.
+    """
+
+    resource = models.CharField(_("Resource"), max_length=50, unique=True)
+    last_updated_at = models.DateTimeField(_("Last updated at"), null=True, blank=True)
+    last_success_at = models.DateTimeField(_("Last success at"), null=True, blank=True)
+
+    class Meta:
+        verbose_name = _("Core sync state")
+        verbose_name_plural = _("Core sync states")
+
+    def __unicode__(self):
+        return self.resource
+
+    def __str__(self):
+        return self.resource
+
+    @classmethod
+    def get_for_resource(cls, resource):
+        obj, _ = cls.objects.get_or_create(resource=resource)
+        return obj
+
+    def get_from_date_updated(self, default):
+        """
+        Retorna a data inicial para o filtro ``from_date_created`` da API.
+
+        Usa sempre a última data coletada; se ainda não houver checkpoint,
+        retorna ``default``.
+        """
+        if self.last_updated_at:
+            return self.last_updated_at.date().isoformat()
+        return default
+
+    def update_checkpoint(self, max_updated_at=None):
+        """
+        Atualiza o checkpoint após uma execução bem-sucedida de sync.
+        """
+        update_fields = ["last_success_at"]
+        if max_updated_at:
+            self.last_updated_at = max_updated_at
+            update_fields.append("last_updated_at")
+        self.last_success_at = timezone.now()
+        self.save(update_fields=update_fields)

core/utils/sync_state.py

@@ -0,0 +1,49 @@
+from django.utils import timezone
+from django.utils.dateparse import parse_datetime
+
+
+def _normalize_datetime(value):
+    if value is None:
+        return None
+    if hasattr(value, "utcoffset"):
+        dt = value
+    else:
+        dt = parse_datetime(str(value))
+    if dt is None:
+        return None
+    if timezone.is_naive(dt):
+        dt = timezone.make_aware(dt, timezone.utc)
+    return dt
+
+
+def track_max_from_item(current_max, item, field="updated"):
+    """
+    Retorna o timestamp mais recente encontrado ao iterar resultados da API.
+
+    Converte ``item[field]`` e ``current_max`` para ``datetime`` antes de
+    comparar, evitando erro ao misturar string ISO da API com ``DateTimeField``.
+
+    Args:
+        current_max: ``datetime`` já processado, ou None.
+        item: Dicionário retornado pela API Core.
+        field: Nome do campo de data em ``item`` (padrão: ``created``).
+
+    Returns:
+        O ``datetime`` mais recente entre ``current_max`` e ``item[field]``.
+    """
+    value = _normalize_datetime(item.get(field))
+    current_max = _normalize_datetime(current_max)
+    if value and (current_max is None or value > current_max):
+        return value
+    return current_max
+
+
+def finalize_core_sync_state(sync_state, max_updated_at):
+    """
+    Persiste o checkpoint após uma execução bem-sucedida de sync da API Core.
+
+    Args:
+        sync_state: Instância de ``CoreSyncState`` do recurso sincronizado.
+        max_updated_at: Maior ``created`` (ou equivalente) visto na execução.
+    """
+    sync_state.update_checkpoint(max_updated_at)