misc ddex cleanup (#8368)

dev-tools/config.json

@@ -138,7 +138,7 @@
     "DDEX_KEY": "49d5e13d355709b615b7cce7369174fb240b6b39",
     "DDEX_SECRET": "2b2c2b90d9a489234ae629a5284de84fb0633306257f17667aaebf2345d92152",
     "SESSION_SECRET": "something random",
-    "DDEX_ADMIN_ALLOWLIST": "127559427",
+    "DDEX_ADMIN_ALLOWLIST": "127559427,555164012",
     "NODE_ENV": "stage"
   }
 }

packages/ddex/ingester/README.md

@@ -5,7 +5,7 @@ Crawls and parses new DDEX uploads.
 ### Local Dev
 1. Make sure the DDEX dependencies are running: `audius-compose up --ddex-deps`
 2. (Optional) See the webapp README to start that server and go through the OAuth flow with a staging user
-3. Parse a file: `IS_DEV=true AWS_ENDPOINT=http://ingress:4566 DDEX_CHOREOGRAPHY=ERNBatched go run cmd/main.go ./e2e_test/fixtures/batch/fuga/20240305090456555 --wipe`
+3. Parse a file: `IS_DEV=true AWS_ENDPOINT=http://ingress:4566 DDEX_CHOREOGRAPHY=ERNBatched go run cmd/main.go ./e2e_test/fixtures/batch/fuga/20240305090206405 --wipe`
 4. Alternatively, run `IS_DEV=true AWS_ENDPOINT=http://ingress:4566 DDEX_CHOREOGRAPHY=ERNBatched air` to run the server with hot reloading. Then, run the webapp (see its README) to use its "Re-process All" button to test code changes against files you sync using the below instructions
 
 
@@ -19,4 +19,4 @@ Usage:
 go run ./cmd/sync s3://ddex-prod-<provider>-raw/20240305090456555
 
 aws s3 ls s3://ddex-prod-<provider>-raw/20240305090456555 --profile local
-```
+```

packages/ddex/ingester/parser/ern38x.go

@@ -1144,7 +1144,6 @@ func processSoundRecordingNode(sNode *xmlquery.Node) (recording *SoundRecording,
 		name := safeInnerText(contributorNode.SelectElement("PartyName/FullName"))
 		seqNo, seqNoErr := strconv.Atoi(contributorNode.SelectAttr("SequenceNumber"))
 		if seqNoErr != nil {
-			fmt.Printf("error parsing ResourceContributor %s's SequenceNumber: %v\n", name, seqNoErr)
 			seqNo = -1
 		}
 		contributor := common.ResourceContributor{
@@ -1168,7 +1167,6 @@ func processSoundRecordingNode(sNode *xmlquery.Node) (recording *SoundRecording,
 		name := safeInnerText(indirectContributorNode.SelectElement("PartyName/FullName"))
 		seqNo, seqNoErr := strconv.Atoi(indirectContributorNode.SelectAttr("SequenceNumber"))
 		if seqNoErr != nil {
-			fmt.Printf("error parsing IndirectResourceContributor %s's SequenceNumber: %v\n", name, seqNoErr)
 			seqNo = -1
 		}
 		contributor := common.ResourceContributor{

packages/discovery-provider/src/tasks/entity_manager/entities/playlist.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Union
 
 from sqlalchemy import desc
@@ -23,6 +23,7 @@
     copy_record,
     is_ddex_signer,
     validate_signer,
+    parse_release_date,
 )
 from src.tasks.metadata import immutable_playlist_fields
 from src.tasks.task_helpers import generate_slug_and_collision_id
@@ -312,6 +313,12 @@ def validate_playlist_tx(params: ManageEntityParameters):
             raise IndexingValidationError(
                 f"Cannot create playlist {playlist_id} below the offset"
             )
+        if is_ddex_signer(params.signer):
+            parsed_release_date = parse_release_date(params.metadata["release_date"])
+            if parsed_release_date and parsed_release_date > datetime.now().astimezone(timezone.utc):
+                raise IndexingValidationError(
+                    f"Cannot create playlist {playlist_id} with a future relaese date"
+                )
     else:
         if playlist_id not in params.existing_records["Playlist"]:
             raise IndexingValidationError(
@@ -437,8 +444,13 @@ def create_playlist(params: ManageEntityParameters):
     last_added_to = None
 
     ddex_app = None
+    created_at = params.block_datetime
     if is_ddex_signer(params.signer):
         ddex_app = params.signer
+        if params.action == Action.CREATE:
+            parsed_release_date = parse_release_date(params.metadata["release_date"])
+            if parsed_release_date:
+                created_at = str(parsed_release_date)  # type: ignore
 
     for track in tracks:
         if "track" not in track or "time" not in track:
@@ -471,7 +483,7 @@ def create_playlist(params: ManageEntityParameters):
         is_stream_gated=params.metadata.get("is_stream_gated", False),
         stream_conditions=params.metadata.get("stream_conditions", None),
         playlist_contents={"track_ids": tracks_with_index_time},
-        created_at=params.block_datetime,
+        created_at=created_at,
         updated_at=params.block_datetime,
         blocknumber=params.block_number,
         blockhash=params.event_blockhash,
@@ -667,6 +679,9 @@ def process_playlist_data_event(
     playlist_record.updated_at = block_datetime
     playlist_record.metadata_multihash = metadata_cid
 
+    if is_ddex_signer(params.signer):
+        playlist_record.ddex_app = params.signer
+
     params.logger.info(
         f"playlist.py | EntityManager | Updated playlist record {playlist_record}"
     )

packages/discovery-provider/src/tasks/entity_manager/entities/track.py

@@ -29,6 +29,7 @@
     copy_record,
     is_ddex_signer,
     validate_signer,
+    parse_release_date,
 )
 from src.tasks.metadata import immutable_track_fields
 from src.tasks.task_helpers import generate_slug_and_collision_id
@@ -229,33 +230,6 @@ def is_valid_json_field(metadata, field):
     return False
 
 
-def parse_release_date(release_date_str):
-    # try various time formats
-    if not release_date_str:
-        return None
-
-    try:
-        return datetime.strptime(
-            release_date_str, "%a %b %d %Y %H:%M:%S GMT%z"
-        ).astimezone(timezone.utc)
-    except ValueError:
-        pass
-
-    try:
-        return datetime.strptime(release_date_str, "%Y-%m-%dT%H:%M:%S.%fZ").astimezone(
-            timezone.utc
-        )
-    except ValueError:
-        pass
-
-    try:
-        return datetime.fromtimestamp(int(release_date_str)).astimezone(timezone.utc)
-    except (ValueError, TypeError):
-        pass
-
-    return None
-
-
 def populate_track_record_metadata(track_record: Track, track_metadata, handle, action):
     # Iterate over the track_record keys
     # Update track_record values for which keys exist in track_metadata
@@ -510,6 +484,9 @@ def update_track_record(
 ):
     populate_track_record_metadata(track, metadata, handle, params.action)
 
+    if is_ddex_signer(params.signer):
+        track.ddex_app = params.signer
+
     # if cover_art CID is of a dir, store under _sizes field instead
     if track.cover_art:
         track.cover_art_sizes = track.cover_art

packages/discovery-provider/src/tasks/entity_manager/utils.py

@@ -1,6 +1,6 @@
 import json
 import os
-from datetime import datetime
+from datetime import datetime, timezone
 from enum import Enum
 from typing import Dict, List, Literal, Set, Tuple, TypedDict, Union
 
@@ -512,3 +512,30 @@ def is_ddex_signer(signer):
             address.lower() for address in ddex_apps.split(",")
         )
     return False
+
+
+def parse_release_date(release_date_str):
+    # try various time formats
+    if not release_date_str:
+        return None
+
+    try:
+        return datetime.strptime(
+            release_date_str, "%a %b %d %Y %H:%M:%S GMT%z"
+        ).astimezone(timezone.utc)
+    except ValueError:
+        pass
+
+    try:
+        return datetime.strptime(release_date_str, "%Y-%m-%dT%H:%M:%S.%fZ").astimezone(
+            timezone.utc
+        )
+    except ValueError:
+        pass
+
+    try:
+        return datetime.fromtimestamp(int(release_date_str)).astimezone(timezone.utc)
+    except (ValueError, TypeError):
+        pass
+
+    return None