feat: add ArtistContentConsistencyScorer transformer

test/test_transformers/test_artist_content_consistency_scorer.py

@@ -0,0 +1,167 @@
+from pyspark.sql.types import (
+    ArrayType,
+    FloatType,
+    IntegerType,
+    StringType,
+    StructField,
+    StructType,
+)
+
+from test.pyspark_test import PySparkTest
+from tidal_per_transformers.transformers.artist_content_consistency_scorer import (
+    ArtistContentConsistencyScorer,
+)
+
+
+class TestArtistContentConsistencyScorer(PySparkTest):
+
+    TRACK_SCHEMA = StructType([
+        StructField("artistId", StringType(), True),
+        StructField("trackGroup", StringType(), True),
+        StructField("copyright", StringType(), True),
+        StructField("isrc", StringType(), True),
+        StructField("duration", IntegerType(), True),
+        StructField("provider", StringType(), True),
+    ])
+
+    ARTIST_METADATA_SCHEMA = StructType([
+        StructField("artistId", StringType(), True),
+        StructField("modalCopyright", StringType(), True),
+        StructField("isrcPrefixes", ArrayType(StringType()), True),
+        StructField("meanDuration", FloatType(), True),
+        StructField("stdDuration", FloatType(), True),
+        StructField("primaryProvider", StringType(), True),
+    ])
+
+    def _create_artist_metadata(self):
+        data = [
+            (
+                "artist_carlos_santana",
+                "Sony Music Entertainment",
+                ["USSM1", "USSM2"],
+                240.0,
+                30.0,
+                "Sony Music",
+            ),
+        ]
+        return self.spark.createDataFrame(data, self.ARTIST_METADATA_SCHEMA)
+
+    def test_legitimate_track_scores_high(self):
+        """All signals match - score should be ~1.0."""
+        tracks = self.spark.createDataFrame(
+            [(
+                "artist_carlos_santana",
+                "track_001",
+                "Sony Music Entertainment",
+                "USSM1234567",
+                235,
+                "Sony Music",
+            )],
+            self.TRACK_SCHEMA,
+        )
+        artist_metadata = self._create_artist_metadata()
+
+        transformer = ArtistContentConsistencyScorer(artist_metadata=artist_metadata)
+        result = transformer.transform(tracks)
+
+        row = result.collect()[0]
+        self.assertAlmostEqual(row["consistencyScore"], 1.0, places=2)
+
+    def test_wrong_artist_scores_low(self):
+        """Santana rapper on Carlos Santana - copyright, ISRC, provider all mismatch."""
+        tracks = self.spark.createDataFrame(
+            [(
+                "artist_carlos_santana",
+                "track_002",
+                "Jive Records",
+                "USJI50012345",
+                220,
+                "Zomba Recording",
+            )],
+            self.TRACK_SCHEMA,
+        )
+        artist_metadata = self._create_artist_metadata()
+
+        transformer = ArtistContentConsistencyScorer(artist_metadata=artist_metadata)
+        result = transformer.transform(tracks)
+
+        row = result.collect()[0]
+        # Only duration normality contributes (220 is within 1 std of 240)
+        # Expected: 0.15 * 1.0 = 0.15
+        self.assertAlmostEqual(row["consistencyScore"], 0.15, places=2)
+
+    def test_test_upload_scores_very_low(self):
+        """Kokubo 'Do Re Mi test' - copyright mismatch, ISRC mismatch, very short duration."""
+        tracks = self.spark.createDataFrame(
+            [(
+                "artist_carlos_santana",
+                "track_003",
+                "Independent Upload",
+                "GBXXX000001",
+                15,
+                "TuneCore",
+            )],
+            self.TRACK_SCHEMA,
+        )
+        artist_metadata = self._create_artist_metadata()
+
+        transformer = ArtistContentConsistencyScorer(artist_metadata=artist_metadata)
+        result = transformer.transform(tracks)
+
+        row = result.collect()[0]
+        # Duration z-score: |15 - 240| / 30 = 7.5 -> beyond 3 std devs -> 0.0
+        # All signals 0.0 -> score 0.0
+        self.assertAlmostEqual(row["consistencyScore"], 0.0, places=2)
+
+    def test_legitimate_reissue_scores_moderate(self):
+        """Copyright different but ISRC and provider match - reissue scenario."""
+        tracks = self.spark.createDataFrame(
+            [(
+                "artist_carlos_santana",
+                "track_004",
+                "Legacy Recordings",
+                "USSM1999888",
+                250,
+                "Sony Music",
+            )],
+            self.TRACK_SCHEMA,
+        )
+        artist_metadata = self._create_artist_metadata()
+
+        transformer = ArtistContentConsistencyScorer(artist_metadata=artist_metadata)
+        result = transformer.transform(tracks)
+
+        row = result.collect()[0]
+        # Copyright: 0.0 (mismatch) -> 0.35 * 0.0 = 0.0
+        # ISRC: 1.0 (USSM1 in prefixes) -> 0.25 * 1.0 = 0.25
+        # Provider: 1.0 (match) -> 0.25 * 1.0 = 0.25
+        # Duration: z = |250-240|/30 = 0.33 -> within 1 std -> 1.0 -> 0.15 * 1.0 = 0.15
+        # Total: 0.65
+        self.assertAlmostEqual(row["consistencyScore"], 0.65, places=2)
+
+    def test_null_handling_does_not_crash(self):
+        """Missing copyright or ISRC should score 0.0 for that signal, not crash."""
+        tracks = self.spark.createDataFrame(
+            [(
+                "artist_carlos_santana",
+                "track_005",
+                None,
+                None,
+                230,
+                "Sony Music",
+            )],
+            self.TRACK_SCHEMA,
+        )
+        artist_metadata = self._create_artist_metadata()
+
+        transformer = ArtistContentConsistencyScorer(artist_metadata=artist_metadata)
+        result = transformer.transform(tracks)
+
+        row = result.collect()[0]
+        # Copyright: 0.0 (null) -> 0.35 * 0.0 = 0.0
+        # ISRC: 0.0 (null) -> 0.25 * 0.0 = 0.0
+        # Provider: 1.0 (match) -> 0.25 * 1.0 = 0.25
+        # Duration: z = |230-240|/30 = 0.33 -> within 1 std -> 1.0 -> 0.15 * 1.0 = 0.15
+        # Total: 0.40
+        self.assertAlmostEqual(row["consistencyScore"], 0.40, places=2)
+        self.assertEqual(result.count(), 1)

tidal_per_transformers/transformers/__init__.py

@@ -1,6 +1,7 @@
 from .loggable_transformer import LoggableTransformer
 from .aggregate_transformer import AggregateTransformer
 from .artist_compound_transformer import ArtistCompoundMappingTransformer
+from .artist_content_consistency_scorer import ArtistContentConsistencyScorer
 from .artist_filter_transformer import ArtistFilterTransformer
 from .artists_struct_filter_transformer import ArtistsStructFilterTransformer
 from .blazingtext_input_format_transformer import BlazingTextInputFormatTransformer

tidal_per_transformers/transformers/artist_content_consistency_scorer.py

@@ -0,0 +1,94 @@
+from pyspark.sql import DataFrame
+import pyspark.sql.functions as F
+from pyspark.sql.types import FloatType
+
+from tidal_per_transformers.transformers.loggable_transformer import LoggableTransformer
+
+
+class ArtistContentConsistencyScorer(LoggableTransformer):
+    """Scores how likely a track-artist pairing is correct based on metadata consistency.
+
+    Computes a weighted consistency score (0.0 to 1.0) by comparing track-level metadata
+    against pre-computed artist-level aggregates across four signals:
+    - Copyright match (weight 0.35)
+    - ISRC prefix match (weight 0.25)
+    - Provider match (weight 0.25)
+    - Duration normality (weight 0.15)
+    """
+
+    COPYRIGHT_WEIGHT = 0.35
+    ISRC_PREFIX_WEIGHT = 0.25
+    PROVIDER_WEIGHT = 0.25
+    DURATION_WEIGHT = 0.15
+
+    def __init__(self, artist_metadata: DataFrame):
+        """Initialise the scorer with artist-level metadata.
+
+        :param artist_metadata: DataFrame with columns artistId, modalCopyright,
+            isrcPrefixes, meanDuration, stdDuration, primaryProvider
+        """
+        super().__init__()
+        self.artist_metadata = artist_metadata
+
+    def _transform(self, df: DataFrame) -> DataFrame:
+        # Broadcast join artist metadata onto track data
+        joined = df.join(
+            F.broadcast(self.artist_metadata),
+            on="artistId",
+            how="left",
+        )
+
+        # Copyright match: 1.0 if track copyright == modal copyright, else 0.0
+        copyright_score = F.when(
+            F.col("copyright").isNull() | F.col("modalCopyright").isNull(), 0.0
+        ).when(
+            F.col("copyright") == F.col("modalCopyright"), 1.0
+        ).otherwise(0.0).cast(FloatType())
+
+        # ISRC prefix match: 1.0 if first 5 chars of ISRC in artist's prefix set, else 0.0
+        isrc_prefix = F.substring(F.col("isrc"), 1, 5)
+        isrc_score = F.when(
+            F.col("isrc").isNull() | F.col("isrcPrefixes").isNull(), 0.0
+        ).when(
+            F.array_contains(F.col("isrcPrefixes"), isrc_prefix), 1.0
+        ).otherwise(0.0).cast(FloatType())
+
+        # Provider match: 1.0 if track provider == primary provider, else 0.0
+        provider_score = F.when(
+            F.col("provider").isNull() | F.col("primaryProvider").isNull(), 0.0
+        ).when(
+            F.col("provider") == F.col("primaryProvider"), 1.0
+        ).otherwise(0.0).cast(FloatType())
+
+        # Duration normality: 1.0 within 1 std dev, linear decay to 0.0 at 3 std devs
+        z_score = F.abs(
+            (F.col("duration").cast("float") - F.col("meanDuration")) / F.col("stdDuration")
+        )
+        duration_score = F.when(
+            F.col("duration").isNull()
+            | F.col("meanDuration").isNull()
+            | F.col("stdDuration").isNull()
+            | (F.col("stdDuration") == 0.0),
+            0.0,
+        ).when(
+            z_score <= 1.0, 1.0
+        ).when(
+            z_score >= 3.0, 0.0
+        ).otherwise(
+            # Linear decay: (3 - z) / (3 - 1) = (3 - z) / 2
+            (F.lit(3.0) - z_score) / F.lit(2.0)
+        ).cast(FloatType())
+
+        # Weighted sum
+        consistency_score = (
+            F.lit(self.COPYRIGHT_WEIGHT) * copyright_score
+            + F.lit(self.ISRC_PREFIX_WEIGHT) * isrc_score
+            + F.lit(self.PROVIDER_WEIGHT) * provider_score
+            + F.lit(self.DURATION_WEIGHT) * duration_score
+        ).cast(FloatType())
+
+        # Select original columns plus new score, drop joined metadata columns
+        original_columns = df.columns
+        result = joined.withColumn("consistencyScore", consistency_score)
+
+        return result.select(*original_columns, "consistencyScore")