Merge pull request #1498 from DDMAL/develop

Merge develop into staging, 3 Jun 2024

cron/management/manage.sh

@@ -7,4 +7,4 @@
 DOCKER_COMPOSE_FILE=$1                                              # This is the path to the docker-compose file.
 COMMAND=$2                                                          # This is the command to execute.
 
-/usr/local/bin/docker-compose -f $DOCKER_COMPOSE_FILE exec -T django python manage.py $COMMAND
+/usr/local/bin/docker compose -f $DOCKER_COMPOSE_FILE exec -T django python manage.py $COMMAND

cron/postgres/db_backup.sh

@@ -20,9 +20,9 @@ BACKUP_FILENAME=$(date "+%Y-%m-%dT%H:%M:%S").sql.gz                 # This is th
 
 # Create the backup and copy it to the daily backup directory
 mkdir -p $BACKUP_DIR/daily $BACKUP_DIR/weekly $BACKUP_DIR/monthly $BACKUP_DIR/yearly
-/usr/bin/docker exec cantusdb_postgres_1 /usr/local/bin/postgres_backup.sh $BACKUP_FILENAME
-/usr/bin/docker cp cantusdb_postgres_1:/var/lib/postgresql/backups/$BACKUP_FILENAME $BACKUP_DIR/daily
-/usr/bin/docker exec cantusdb_postgres_1 rm /var/lib/postgresql/backups/$BACKUP_FILENAME
+/usr/bin/docker exec cantusdb-postgres-1 /usr/local/bin/postgres_backup.sh $BACKUP_FILENAME
+/usr/bin/docker cp cantusdb-postgres-1:/var/lib/postgresql/backups/$BACKUP_FILENAME $BACKUP_DIR/daily
+/usr/bin/docker exec cantusdb-postgres-1 rm /var/lib/postgresql/backups/$BACKUP_FILENAME
 
 # Manage retention of daily backups
 FILES_TO_REMOVE=$(ls -td $BACKUP_DIR/daily/* | tail -n +8)

django/cantusdb_project/cantusindex.py

@@ -6,6 +6,7 @@
 import requests
 from typing import Optional, Union, Callable
 from main_app.models import Genre
+import json
 
 CANTUS_INDEX_DOMAIN: str = "https://cantusindex.uwaterloo.ca"
 DEFAULT_TIMEOUT: float = 2  # seconds
@@ -112,7 +113,7 @@ def get_suggested_chant(
     }
 
 
-def get_suggested_fulltext(cantus_id: str) -> str:
+def get_suggested_fulltext(cantus_id: str) -> Optional[str]:
     endpoint_path: str = f"/json-cid/{cantus_id}"
     json: Union[dict, list, None] = get_json_from_ci_api(endpoint_path)
 
@@ -128,6 +129,41 @@ def get_suggested_fulltext(cantus_id: str) -> str:
     return suggested_fulltext
 
 
+def get_merged_cantus_ids() -> Optional[list]:
+    """Retrieve merged Cantus IDs from the Cantus Index API (/json-merged-chants)
+
+    This function sends a request to the Cantus Index API endpoint for merged chants
+    and retrieves the response. The response is expected to be a list of dictionaries,
+    each containing information about a merged Cantus ID, including the old Cantus ID,
+    the new Cantus ID, and the date of the merge.
+
+    Returns:
+        Optional[list]: A list of dictionaries representing merged chant information,
+    or None if there was an error retrieving the data or the response format is invalid.
+
+    """
+    endpoint_path: str = "/json-merged-chants"
+
+    # We have to use the old CI domain since the API is still not available on
+    # cantusindex.uwaterloo.ca. Once it's available, we can use get_json_from_ci_api
+    # json: Union[dict, list, None] = get_json_from_ci_api(endpoint_path)
+    uri: str = f"https://cantusindex.org{endpoint_path}"
+    try:
+        response: requests.Response = requests.get(uri, timeout=DEFAULT_TIMEOUT)
+    except requests.exceptions.Timeout:
+        return None
+    if not response.status_code == 200:
+        return None
+    response.encoding = "utf-8-sig"
+    raw_text: str = response.text
+    text_without_bom: str = raw_text.encode().decode("utf-8-sig")
+    merge_events: list = json.loads(text_without_bom)
+
+    if not isinstance(merge_events, list):
+        return None
+    return merge_events
+
+
 def get_json_from_ci_api(
     path: str, timeout: float = DEFAULT_TIMEOUT
 ) -> Union[dict, list, None]:

django/cantusdb_project/main_app/management/commands/add_cantus_index_merge_events.py

@@ -0,0 +1,96 @@
+from typing import Optional
+from datetime import datetime
+from django.core.management.base import BaseCommand
+from main_app.models import Chant
+from cantusindex import get_merged_cantus_ids
+
+
+class Command(BaseCommand):
+    help = (
+        "Fetch and apply merge events from the /json-merged-chants API on Cantus Index"
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "date",
+            nargs="?",
+            type=str,
+            help="Filter merges by date in the format YYYY-MM-DD",
+        )
+
+    def handle(self, *args, **kwargs):
+        date_filter: str = kwargs["date"]
+        if date_filter:
+            try:
+                date_filter = datetime.strptime(date_filter, "%Y-%m-%d")
+            except ValueError:
+                self.stdout.write(
+                    self.style.ERROR("Invalid date format. Please use YYYY-MM-DD.")
+                )
+                return
+
+        # Fetch the merge events from the Cantus Index API
+        merge_events: Optional[list] = get_merged_cantus_ids()
+
+        if not merge_events:
+            self.stdout.write(
+                self.style.ERROR("Failed to fetch data from Cantus Index.")
+            )
+            return
+
+        # Filter merge events by the provided date
+        if date_filter:
+            merge_events = self.filter_merge_events_by_date(merge_events, date_filter)
+
+        # Apply the new merge events on Cantus DB
+        for tx in merge_events:
+            self.apply_transaction(tx)
+
+    def filter_merge_events_by_date(
+        self, merge_events: list, date_filter: datetime
+    ) -> list:
+        filtered_merge_events = []
+        for tx in merge_events:
+            try:
+                tx_date = datetime.strptime(tx["date"], "%Y-%m-%d")
+                if tx_date > date_filter:
+                    filtered_merge_events.append(tx)
+            except ValueError:
+                # Handle invalid date format (usually 0000-00-00)
+                self.stdout.write(
+                    self.style.WARNING(
+                        f"Ignoring merge event with invalid date format: {tx['date']}"
+                    )
+                )
+        return filtered_merge_events
+
+    def apply_transaction(self, transaction: dict) -> None:
+        old_cantus_id: str = transaction["old"]
+        new_cantus_id: str = transaction["new"]
+        if not old_cantus_id or not new_cantus_id:
+            self.stdout.write(
+                self.style.WARNING(
+                    f"Skipping transaction with missing cantus_id. Old: {old_cantus_id}, New: {new_cantus_id}"
+                )
+            )
+            return
+
+        affected_chants = Chant.objects.filter(cantus_id=old_cantus_id)
+        if affected_chants:
+            try:
+                affected_chants.update(cantus_id=new_cantus_id)
+                self.stdout.write(
+                    self.style.SUCCESS(
+                        f"Old Cantus ID: {old_cantus_id} -> New Cantus ID: {new_cantus_id}\n"
+                    )
+                )
+            except Exception as e:
+                self.stdout.write(
+                    self.style.ERROR(
+                        f"An error occurred while updating Chants with old cantus_id {old_cantus_id}: {e}"
+                    )
+                )
+        else:
+            self.stdout.write(
+                self.style.WARNING(f"No Chants found with cantus_id: {old_cantus_id}")
+            )

django/cantusdb_project/main_app/management/commands/assign_chants_to_segments.py

@@ -7,7 +7,7 @@
 """
 
 from django.core.management.base import BaseCommand
-from main_app.models import Source, Chant, Segment
+from main_app.models import Source, Chant, Segment, Sequence
 
 
 class Command(BaseCommand):
@@ -18,4 +18,27 @@ def handle(self, *args, **options):
         for source in sources:
             segment = Segment.objects.get(id=source.segment_id)
             chants = Chant.objects.filter(source=source)
-            chants.update(segment=segment)
+            sequences = Sequence.objects.filter(source=source)
+            chants_count = chants.count()
+            sequences_count = sequences.count()
+            if chants_count != 0 and sequences_count != 0:
+                self.stdout.write(
+                    self.style.ERROR(
+                        f"Source {source.id} has {chants_count} chants and {sequences_count} sequences."
+                    )
+                )
+                continue
+            if chants_count > 0:
+                chants.update(segment=segment)
+                self.stdout.write(
+                    self.style.SUCCESS(
+                        f"Assigned {chants_count} chants in source {source.id} to segment {segment.id}."
+                    )
+                )
+            else:
+                sequences.update(segment=segment)
+                self.stdout.write(
+                    self.style.SUCCESS(
+                        f"Assigned {sequences_count} sequences in source {source.id} to segment {segment.id}."
+                    )
+                )

django/cantusdb_project/main_app/templates/chant_detail.html

@@ -471,7 +471,7 @@ <h4>List of melodies</h4>
                                     </ul>
                                 {% endif %}
 
-                                {{ source.indexing_notes|default_if_none:"" }}
+                                {{ source.indexing_notes|default_if_none:""|safe }}
                                 <br>
                                 {% with creator=source.created_by %}
                                     {% if creator %}

django/cantusdb_project/main_app/templates/source_detail.html

@@ -289,7 +289,7 @@ <h4>{{ source.siglum }}</h4>
                                 {% endfor %}
                             </ul>
                         {% endif %}
-                        {{ source.indexing_notes|default_if_none:"" }}
+                        {{ source.indexing_notes|default_if_none:""|safe }}
                         <br>
                         {% with creator=source.created_by %}
                             {% if creator %}

django/cantusdb_project/main_app/tests/test_assign_chants_to_segments.py

@@ -1,7 +1,7 @@
 from django.test import TestCase
 from django.core.management import call_command
 
-from main_app.models import Chant
+from main_app.models import Chant, Sequence
 
 from main_app.tests.make_fakes import make_fake_source, make_fake_segment
 
@@ -12,17 +12,26 @@ def test_assign_chants_to_segments(self):
         segment_2 = make_fake_segment()
         source_1 = make_fake_source(segment=segment_1)
         source_2 = make_fake_source(segment=segment_2)
+        sequence_source = make_fake_source(segment=segment_2)
         for _ in range(5):
             Chant.objects.create(source=source_1)
         for _ in range(3):
             Chant.objects.create(source=source_2)
+        for _ in range(4):
+            Sequence.objects.create(source=sequence_source)
         all_chants = Chant.objects.all()
         for chant in all_chants:
             self.assertIsNone(chant.segment_id)
+        all_sequences = Sequence.objects.all()
+        for sequence in all_sequences:
+            self.assertIsNone(sequence.segment_id)
         call_command("assign_chants_to_segments")
         source_1_chants = Chant.objects.filter(source=source_1)
         source_2_chants = Chant.objects.filter(source=source_2)
+        sequence_source_sequences = Sequence.objects.filter()
         for chant in source_1_chants:
             self.assertEqual(chant.segment_id, segment_1.id)
         for chant in source_2_chants:
             self.assertEqual(chant.segment_id, segment_2.id)
+        for sequence in sequence_source_sequences:
+            self.assertEqual(sequence.segment_id, segment_2.id)