From 368fa94f9e32fc6fb7f44b71a17246b2239da4fd Mon Sep 17 00:00:00 2001 From: Jeff Agapitos <233853744+jeffa-block@users.noreply.github.com> Date: Sun, 31 May 2026 22:08:46 +1000 Subject: [PATCH] feat: add ArtistContentConsistencyScorer transformer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scores track-artist pairings by metadata consistency across four signals: - Copyright holder match (weight 0.35) - ISRC prefix match (weight 0.25) - Provider/distributor match (weight 0.25) - Duration normality (weight 0.15) Outputs a consistencyScore column (0.0-1.0) that downstream systems can use to detect content misattribution (e.g. different artists with the same name merged onto one profile, test uploads on real artist pages). This is a read-only scoring transformer with no side effects — it does not filter, hide, or remove any content. It adds a column that other transformers or services can optionally consume. Includes 5 test cases covering: legitimate tracks, wrong-artist merges, test uploads, legitimate reissues, and null handling. --- .../test_artist_content_consistency_scorer.py | 167 ++++++++++++++++++ .../transformers/__init__.py | 1 + .../artist_content_consistency_scorer.py | 94 ++++++++++ 3 files changed, 262 insertions(+) create mode 100644 test/test_transformers/test_artist_content_consistency_scorer.py create mode 100644 tidal_per_transformers/transformers/artist_content_consistency_scorer.py diff --git a/test/test_transformers/test_artist_content_consistency_scorer.py b/test/test_transformers/test_artist_content_consistency_scorer.py new file mode 100644 index 0000000..124e2cc --- /dev/null +++ b/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) diff --git a/tidal_per_transformers/transformers/__init__.py b/tidal_per_transformers/transformers/__init__.py index 4bb4006..3b67317 100644 --- a/tidal_per_transformers/transformers/__init__.py +++ b/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 diff --git a/tidal_per_transformers/transformers/artist_content_consistency_scorer.py b/tidal_per_transformers/transformers/artist_content_consistency_scorer.py new file mode 100644 index 0000000..8cf7ba6 --- /dev/null +++ b/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")