Permite selecionar diretórios de logs por padrão configurável

log_manager/services/catalog.py

@@ -39,12 +39,13 @@ def catalog_log_files_from_configured_directories(
             )
 
         for directory in directories:
-            _catalog_log_files_in_directory(
-                collection=collection,
-                directory_path=directory.path,
-                visible_dates=visible_dates,
-                supported_extensions=supported_extensions,
-            )
+            for path in directory.get_log_directories():
+                _catalog_log_files_in_directory(
+                    collection=collection,
+                    directory_path=path,
+                    visible_dates=visible_dates,
+                    supported_extensions=supported_extensions,
+                )
 
 
 def _catalog_log_files_in_directory(

log_manager/tests/test_catalog.py

@@ -4,27 +4,79 @@
 from pathlib import Path
 from unittest.mock import patch
 
-from django.test import TestCase
+from django.test import TestCase, override_settings
 
 from collection.models import Collection
 from log_manager import choices, file_errors, utils
 from log_manager.models import LogFile
 from log_manager.services import catalog, validation
+from log_manager_config.models import CollectionLogDirectory, LogManagerCollectionConfig
 
 
 class CatalogLogFilesTests(TestCase):
     def setUp(self):
         self.collection = Collection.objects.create(acron3="per", acron2="pe")
 
-    def _catalog_directory(self, directory, visible_dates=None):
-        catalog._catalog_log_files_in_directory(
-            collection=self.collection,
-            directory_path=directory,
-            visible_dates=visible_dates
-            if visible_dates is not None
-            else [date.today()],
-            supported_extensions=[".gz"],
-        )
+    @override_settings(SUPPORTED_LOGFILE_EXTENSIONS=[".gz"])
+    def test_pattern_catalogs_all_logs_and_discovers_new_months(self):
+        config = LogManagerCollectionConfig.objects.create(collection=self.collection)
+
+        with tempfile.TemporaryDirectory() as directory:
+            CollectionLogDirectory.objects.create(
+                config=config,
+                path=directory,
+                subdirectory_pattern="*/*/preprints-scielo-org",
+            )
+            first_month = (
+                Path(directory) / "Y2026" / "September" / "preprints-scielo-org"
+            )
+            other_site = first_month.parent / "another-site"
+            first_month.mkdir(parents=True)
+            other_site.mkdir()
+            paths = [first_month / "access.log.gz", first_month / "error.log.gz"]
+            for path in paths + [other_site / "access.log.gz"]:
+                with gzip.open(path, "wt") as output:
+                    output.write(str(path))
+
+            catalog.catalog_log_files_from_configured_directories(
+                collections=["per"], from_date=date.today(), until_date=date.today()
+            )
+            self.assertEqual(
+                set(LogFile.objects.values_list("path", flat=True)),
+                set(map(str, paths)),
+            )
+
+            next_month = Path(directory) / "Y2026" / "October" / "preprints-scielo-org"
+            next_month.mkdir(parents=True)
+            paths.append(next_month / "access.log.gz")
+            with gzip.open(paths[-1], "wt") as output:
+                output.write("new month")
+
+            catalog.catalog_log_files_from_configured_directories(
+                collections=["per"], from_date=date.today(), until_date=date.today()
+            )
+
+            self.assertEqual(
+                set(LogFile.objects.values_list("path", flat=True)),
+                set(map(str, paths)),
+            )
+
+    @override_settings(SUPPORTED_LOGFILE_EXTENSIONS=[".gz"])
+    def test_blank_pattern_preserves_recursive_cataloging(self):
+        config = LogManagerCollectionConfig.objects.create(collection=self.collection)
+
+        with tempfile.TemporaryDirectory() as directory:
+            CollectionLogDirectory.objects.create(config=config, path=directory)
+            path = Path(directory) / "nested" / "access.log.gz"
+            path.parent.mkdir()
+            with gzip.open(path, "wt") as output:
+                output.write("nested log")
+
+            catalog.catalog_log_files_from_configured_directories(
+                collections=["per"], from_date=date.today(), until_date=date.today()
+            )
+
+            self.assertEqual(LogFile.objects.get().path, str(path))
 
     def test_corrupted_gzip_is_recorded_once_as_error(self):
         with tempfile.TemporaryDirectory() as directory:
@@ -176,3 +228,13 @@ def test_valid_duplicate_removes_the_error_placeholder(self):
 
         self.assertEqual(LogFile.objects.count(), 1)
         self.assertEqual(LogFile.objects.get().hash, expected_hash)
+
+    def _catalog_directory(self, directory, visible_dates=None):
+        catalog._catalog_log_files_in_directory(
+            collection=self.collection,
+            directory_path=directory,
+            visible_dates=visible_dates
+            if visible_dates is not None
+            else [date.today()],
+            supported_extensions=[".gz"],
+        )

log_manager_config/migrations/0007_collectionlogdirectory_subdirectory_pattern.py

@@ -0,0 +1,24 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        (
+            "log_manager_config",
+            "0006_logmanagercollectionconfig_opensearch_index_policy",
+        ),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="collectionlogdirectory",
+            name="subdirectory_pattern",
+            field=models.CharField(
+                blank=True,
+                default="",
+                help_text="Optional pattern relative to Path, such as */*/preprints-scielo-org. Leave blank to scan the entire directory.",
+                max_length=255,
+                verbose_name="Subdirectory Pattern",
+            ),
+        ),
+    ]

log_manager_config/models.py

@@ -1,5 +1,9 @@
 import logging
+import os
+from fnmatch import fnmatchcase
+from glob import iglob
 
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
@@ -140,6 +144,16 @@ class CollectionLogDirectory(Orderable, CommonControlField):
         blank=True,
         null=True,
     )
+    subdirectory_pattern = models.CharField(
+        verbose_name=_("Subdirectory Pattern"),
+        max_length=255,
+        blank=True,
+        default="",
+        help_text=_(
+            "Optional pattern relative to Path, such as */*/preprints-scielo-org. "
+            "Leave blank to scan the entire directory."
+        ),
+    )
     active = models.BooleanField(
         verbose_name=_("Active"),
         default=True,
@@ -154,6 +168,53 @@ class CollectionLogDirectory(Orderable, CommonControlField):
     def __str__(self):
         return f"{self.config.collection} - {self.path} - {self.directory_name}"
 
+    def clean(self):
+        super().clean()
+        pattern = self.subdirectory_pattern
+
+        if os.path.isabs(pattern) or ".." in pattern.split(os.sep):
+            raise ValidationError(
+                {
+                    "subdirectory_pattern": _(
+                        "Use a relative pattern without parent directories."
+                    )
+                }
+            )
+
+    def get_log_directories(self):
+        if not self.subdirectory_pattern:
+            yield self.path
+            return
+
+        for path in iglob(os.path.join(self.path, self.subdirectory_pattern)):
+            if os.path.isdir(path):
+                yield path
+
+    def matches_log_path(self, path):
+        directory_path = os.path.abspath(self.path)
+        file_path = os.path.abspath(path)
+        if os.path.commonpath([directory_path, file_path]) != directory_path:
+            return False
+
+        pattern = os.path.normpath(self.subdirectory_pattern)
+        if pattern == ".":
+            return True
+
+        relative_path = os.path.relpath(os.path.dirname(file_path), directory_path)
+        parts = relative_path.split(os.sep)
+        patterns = pattern.split(os.sep)
+        if len(parts) < len(patterns):
+            return False
+
+        for part, pattern in zip(parts, patterns):
+            if part.startswith(".") and not pattern.startswith("."):
+                return False
+
+            if not fnmatchcase(part, pattern):
+                return False
+
+        return True
+
     @classmethod
     def load(cls, data, user):
         for item in data:
@@ -174,6 +235,7 @@ def load(cls, data, user):
                 path=item.get("path"),
                 active=item.get("directory_active", item.get("active", True)),
                 translator_class=item.get("translator_class", "classic"),
+                subdirectory_pattern=item.get("subdirectory_pattern", ""),
             )
 
     @classmethod
@@ -185,6 +247,7 @@ def create_or_update(
         path,
         active,
         translator_class="classic",
+        subdirectory_pattern="",
     ):
         try:
             obj = cls.objects.get(config=config, path=path)
@@ -200,7 +263,9 @@ def create_or_update(
         obj.path = path
         obj.active = active
         obj.translator_class = translator_class or "classic"
+        obj.subdirectory_pattern = subdirectory_pattern
 
+        obj.clean()
         obj.save()
         logging.info(f"{config.collection.acron3} - {directory_name} - {path}")
         return obj

log_manager_config/tests/test_models.py

@@ -1,3 +1,4 @@
+from django.core.exceptions import ValidationError
 from django.test import TestCase
 
 from collection.models import Collection
@@ -105,6 +106,7 @@ def test_load_uses_directory_active_independently(self):
                     "directory_name": "legacy logs",
                     "path": "/data/logs/legacy",
                     "directory_active": False,
+                    "subdirectory_pattern": "*/*/preprints-scielo-org",
                 }
             ],
             self.user,
@@ -113,3 +115,16 @@ def test_load_uses_directory_active_independently(self):
         directory = CollectionLogDirectory.objects.get(path="/data/logs/legacy")
 
         self.assertFalse(directory.active)
+        self.assertEqual(directory.subdirectory_pattern, "*/*/preprints-scielo-org")
+
+    def test_rejects_patterns_outside_root(self):
+        for pattern in ("/other/logs", "../logs", "*/../logs"):
+            with self.subTest(pattern=pattern):
+                directory = CollectionLogDirectory(
+                    config=self.config,
+                    path="/data/logs",
+                    subdirectory_pattern=pattern,
+                )
+
+                with self.assertRaises(ValidationError):
+                    directory.clean()

metrics/services/parsing/metadata.py

@@ -61,7 +61,7 @@ def _get_log_file_translator_class(log_file):
     for directory in CollectionLogDirectory.objects.filter(
         config__collection=log_file.collection,
     ):
-        if directory.path in log_file.path and directory.translator_class:
+        if directory.translator_class and directory.matches_log_path(log_file.path):
             return _get_translator_class(directory.translator_class)
 
     return None

metrics/tests/parsing/test_metadata.py

@@ -6,6 +6,7 @@
 from collection.models import Collection
 from log_manager_config.models import CollectionLogDirectory, LogManagerCollectionConfig
 from metrics.services.parsing.metadata import (
+    _get_log_file_translator_class,
     _get_translator_class,
     build_url_translation_manager,
 )
@@ -57,6 +58,46 @@ def setUp(self):
             translator_class="books",
         )
 
+    def test_translator_respects_subdirectory_pattern(self):
+        config = self.collection.log_manager_config
+
+        directory = CollectionLogDirectory.objects.create(
+            config=config,
+            path="/unmounted/submission-node01",
+            subdirectory_pattern="*/*/preprints-scielo-org",
+            translator_class="preprints",
+        )
+        log_file = SimpleNamespace(collection=self.collection)
+
+        for suffix in (
+            "Y2026/October/preprints-scielo-org/access.log.gz",
+            "Y2026/October/preprints-scielo-org/nested/error.log.gz",
+        ):
+            with self.subTest(suffix=suffix):
+                log_file.path = f"/unmounted/submission-node01/{suffix}"
+                translator = _get_log_file_translator_class(log_file)
+
+                self.assertEqual(translator.__name__, "URLTranslatorPreprintsSite")
+
+        for path in (
+            "/unmounted/submission-node01/Y2026/October/other-site/access.log.gz",
+            "/unmounted/submission-node01/Y2026/preprints-scielo-org/access.log.gz",
+            "/unmounted/submission-node01/.hidden/October/preprints-scielo-org/access.log.gz",
+            "/unmounted/submission-node010/Y2026/October/preprints-scielo-org/access.log.gz",
+        ):
+            with self.subTest(path=path):
+                log_file.path = path
+
+                self.assertIsNone(_get_log_file_translator_class(log_file))
+
+        directory.subdirectory_pattern = "."
+        directory.save(update_fields=["subdirectory_pattern"])
+        log_file.path = "/unmounted/submission-node01/nested/access.log.gz"
+
+        translator = _get_log_file_translator_class(log_file)
+
+        self.assertEqual(translator.__name__, "URLTranslatorPreprintsSite")
+
     @override_settings(PARSING_METADATA_CACHE_COLLECTIONS=[])
     @patch("metrics.services.parsing.metadata.url_translator.URLTranslationManager")
     @patch("metrics.services.parsing.metadata.Source.metadata")