Import the settled coding into the working store

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tools/src/pop_fem_audit_tools/commands/build_db.py

@@ -7,10 +7,10 @@
 Rebuilds the working store from scratch out of the committed
 inputs: the year-end chart CSV and the output directory for the
 review CSV files, given as the two positional command-line
-arguments, and the optional capture inputs, each given as an
-option: the lyrics cache directory and the Wikidata artist
-snapshot CSV.  An omitted option
-leaves its capture layer unloaded; a given option whose path does
+arguments, and the optional inputs, each given as an
+option: the lyrics cache directory, the Wikidata artist
+snapshot CSV, and the settled coding table CSV.  An omitted option
+leaves its layer unloaded; a given option whose path does
 not exist fails the build.  Missing tables
 are created on a fresh store; existing tables are never altered,
 as the schema lifecycle belongs to the migrations.  Every rebuild
@@ -63,6 +63,7 @@ from ..database import Base, ds
 from ..models import (
     Artist,
     ChartEntry,
+    Coding,
     Role,
     Song,
     SongArtist,
@@ -104,6 +105,9 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
     parser.add_argument(
         "--wikidata-csv", type=Path, default=None,
         help="the Wikidata artist snapshot CSV file to apply")
+    parser.add_argument(
+        "--codings", type=Path, default=None,
+        help="the settled coding table CSV file to import")
     return parser.parse_args(argv)
 
 
@@ -589,6 +593,110 @@ class CaptureImporter:
                         setattr(artist, attribute, row[column])
 
 
+class CodingImporter:
+    """The coding-import job: loads the settled coding table onto
+    the stored songs."""
+
+    COLUMNS: tuple[str, ...] = (
+        "Song", "Artist Credit", "Keyword", "Quote")
+    """The required columns of the coding CSV file."""
+    NEWLINE_ESCAPE: str = "\\n"
+    """The two characters standing for a newline in the quote
+    column, so that the CSV file is one row per line."""
+
+    def __init__(self, session: Session) -> None:
+        """Initialize the importer.
+
+        :param session: The database session.
+        """
+        self.__session: Session = session
+
+    def import_codings(self, path: Path | None) -> None:
+        """Load the settled coding table onto the stored songs.
+
+        A None input leaves the coding unloaded.  Otherwise every
+        row of the CSV file yields one coding of the song named by
+        its title and artist credit, with the quote column stored
+        verbatim except that each two-character ``\\n`` escape
+        becomes a newline; an empty quote column stores an empty
+        string.  When the method returns, the imported codings are
+        queryable in the session.
+
+        :param path: The settled coding table CSV file to import,
+            or None to skip the coding.
+        :return: None.
+        :raises BuildError: When the file lacks a required column,
+            a row names a song that the store does not have, or two
+            rows name the same song and keyword.
+        :raises OSError: When the file cannot be read.
+        """
+        if path is None:
+            return
+        songs: dict[tuple[str, str], Song] = {
+            (x.title, x.artist_credit): x
+            for x in self.__session.scalars(sa.select(Song))}
+        seen: set[tuple[int, str]] = set()
+        with open(path, encoding="utf-8", newline="") as file:
+            reader: csv.DictReader[str] = csv.DictReader(file)
+            self.__check_columns(path, reader.fieldnames)
+            row: dict[str, str]
+            for row in reader:
+                self.__import_coding(path, songs, seen, row)
+        self.__session.flush()
+
+    def __import_coding(self, path: Path,
+                        songs: dict[tuple[str, str], Song],
+                        seen: set[tuple[int, str]],
+                        row: dict[str, str]) -> None:
+        """Store one coding row.
+
+        :param path: The coding CSV file, for the error messages.
+        :param songs: The stored songs, keyed by the title and the
+            artist credit.
+        :param seen: The (song ID, keyword) pairs already stored,
+            updated with the pair of this row.
+        :param row: The coding CSV row.
+        :return: None.
+        :raises BuildError: When the row names a song that the
+            store does not have, or its song and keyword repeat an
+            earlier row.
+        """
+        key: tuple[str, str] = (row["Song"], row["Artist Credit"])
+        song: Song | None = songs.get(key)
+        if song is None:
+            raise BuildError(
+                f"{path}: no song \"{row['Song']}\" by"
+                f" \"{row['Artist Credit']}\"")
+        coding_key: tuple[int, str] = (song.id, row["Keyword"])
+        if coding_key in seen:
+            raise BuildError(
+                f"{path}: duplicated coding: \"{row['Song']}\" by"
+                f" \"{row['Artist Credit']}\", keyword"
+                f" \"{row['Keyword']}\"")
+        seen.add(coding_key)
+        self.__session.add(Coding(
+            song=song, keyword=row["Keyword"],
+            quotes=row["Quote"].replace(self.NEWLINE_ESCAPE, "\n")))
+
+    @classmethod
+    def __check_columns(cls, path: Path,
+                        fieldnames: Sequence[str] | None) -> None:
+        """Verify the coding CSV file has the required columns.
+
+        :param path: The coding CSV file.
+        :param fieldnames: The header row of the file, or None when
+            the file is empty.
+        :return: None.
+        :raises BuildError: When a required column is absent.
+        """
+        header: Sequence[str] = fieldnames or ()
+        missing: list[str] = [
+            x for x in cls.COLUMNS if x not in header]
+        if len(missing) > 0:
+            raise BuildError(
+                f"{path}: missing column(s): {', '.join(missing)}")
+
+
 @dataclass
 class StoreCounts:
     """The row counts of the working store, for the build summary."""
@@ -603,6 +711,8 @@ class StoreCounts:
     """The number of the song-artist credits."""
     songs_with_lyrics: int
     """The number of the songs with lyrics."""
+    codings: int
+    """The number of the settled codings."""
 
     @classmethod
     def get_instance(cls, session: Session) -> Self:
@@ -630,7 +740,9 @@ class StoreCounts:
                 .select_from(SongArtist)),
             songs_with_lyrics=count(
                 sa.select(sa.func.count()).select_from(Song)
-                .where(Song.lyrics.is_not(None))))
+                .where(Song.lyrics.is_not(None))),
+            codings=count(
+                sa.select(sa.func.count()).select_from(Coding)))
 
 
 def reset_store(session: Session) -> None:
@@ -640,7 +752,7 @@ def reset_store(session: Session) -> None:
     :return: None.
     """
     model: type[Base]
-    for model in (SongArtist, ChartEntry, Song, Artist):
+    for model in (Coding, SongArtist, ChartEntry, Song, Artist):
         session.execute(sa.delete(model))
 
 
@@ -811,6 +923,7 @@ def main(argv: list[str] | None = None) -> int:
         ArtistImporter(session).import_artists()
         CaptureImporter(session).import_captures(
             args.lyrics_dir, args.wikidata_csv)
+        CodingImporter(session).import_codings(args.codings)
         counts = StoreCounts.get_instance(session)
         CSVExporter(session, args.derived_dir).write()
         session.commit()
@@ -824,6 +937,7 @@ def main(argv: list[str] | None = None) -> int:
           f" {counts.chart_entries} chart entries,"
           f" {counts.artists} artists,"
           f" {counts.credits} credits,"
-          f" {counts.songs_with_lyrics} songs with lyrics",
+          f" {counts.songs_with_lyrics} songs with lyrics,"
+          f" {counts.codings} codings",
           file=sys.stderr)
     return 0

tools/src/pop_fem_audit_tools/models.py

@@ -6,7 +6,9 @@
 
 The schema covers the year-end chart data: songs with their
 lyrics, their yearly chart entries, the individual artists, and
-the song-artist credits with the role and order.
+the song-artist credits with the role and order.  It also covers
+the settled coding of the songs: the keywords assigned to each
+song with the lyric quotes they are grounded in.
 
 """
 import enum
@@ -45,6 +47,9 @@ class Song(Base):
     song_artists: Mapped[list[SongArtist]] \
         = relationship(back_populates="song")
     """The song-artist credits of the song."""
+    codings: Mapped[list[Coding]] \
+        = relationship(back_populates="song")
+    """The settled codings of the song."""
     __table_args__ = (sa.UniqueConstraint(title, artist_credit),)
     """The table-level constraints."""
 
@@ -112,3 +117,20 @@ class SongArtist(Base):
         sa.CheckConstraint(role.in_([x.value for x in Role]),
                            name="ck_song_artists_role"),)
     """The table-level constraints."""
+
+
+class Coding(Base):
+    """A settled coding keyword of a song, with its lyric quotes."""
+    __tablename__ = "codings"
+    """The table name."""
+
+    song_id: Mapped[int] = mapped_column(sa.ForeignKey(Song.id),
+                                         primary_key=True)
+    """The ID of the coded song."""
+    keyword: Mapped[str] = mapped_column(primary_key=True)
+    """The coding keyword assigned to the song."""
+    quotes: Mapped[str] = mapped_column()
+    """The lyric quotes the keyword is grounded in, joined by a
+    single "|", empty when the keyword carries no evidence."""
+    song: Mapped[Song] = relationship(back_populates="codings")
+    """The coded song."""

tools/tests/test_build_db.py

@@ -21,6 +21,7 @@ from pop_fem_audit_tools.database import DataSource
 from pop_fem_audit_tools.models import (
     Artist,
     ChartEntry,
+    Coding,
     Role,
     Song,
 )
@@ -266,6 +267,7 @@ class TestBuildDB(unittest.TestCase):
         self.__lyrics: Path = self.__dir / "lyrics"
         self.__wikidata: Path = \
             self.__dir / "artists_wikidata.csv"
+        self.__codings: Path = self.__dir / "codings.csv"
         self.__write_chart(self.CHART_CSV)
         url: str = f"sqlite:///{self.__dir}/store.sqlite3"
         config.set_settings(config.Settings(
@@ -290,6 +292,14 @@ class TestBuildDB(unittest.TestCase):
         """
         self.__chart.write_text(content, encoding="utf-8")
 
+    def __write_codings(self, content: str) -> None:
+        """Write the coding CSV fixture.
+
+        :param content: The CSV content.
+        :return: None.
+        """
+        self.__codings.write_text(content, encoding="utf-8")
+
     def __run_build(self, *options: str) -> tuple[int, str]:
         """Run the build with the standard error captured.
 
@@ -821,3 +831,143 @@ class TestBuildDB(unittest.TestCase):
             (self.__derived / "artists.csv").read_text(
                 encoding="utf-8"),
             artists_before)
+
+    CODINGS_CSV: str = (
+        "Song,Artist Credit,Keyword,Quote\n"
+        "Hello,Adele,longing,\"Hello from the other side\\nI must"
+        " have called a thousand times\"\n"
+        "Hello,Adele,regret,I'm sorry|It's no secret\n"
+        "One Dance,Drake featuring Wizkid,desire,\n")
+    """The coding CSV fixture: a two-line quote, two quotes joined
+    by "|", and an empty quote."""
+
+    def __stored_codings(self) -> dict[tuple[str, str], str]:
+        """Read the stored codings keyed by the song and keyword.
+
+        :return: The stored quotes, keyed by the song title and the
+            keyword.
+        """
+        session: Session = self.__session()
+        return {(x.song.title, x.keyword): x.quotes
+                for x in session.scalars(sa.select(Coding))}
+
+    def test_codings_imported(self) -> None:
+        """Test that the coding CSV imports one row per song and
+        keyword, restoring the newline and keeping the "|"-joined
+        quotes verbatim."""
+        self.__write_codings(self.CODINGS_CSV)
+        status: int
+        stderr: str
+        status, stderr = self.__run_build(
+            "--codings", str(self.__codings))
+        self.assertEqual(status, 0)
+        self.assertIn("3 codings", stderr)
+        self.assertEqual(
+            self.__stored_codings(),
+            {("Hello", "longing"):
+                "Hello from the other side\n"
+                "I must have called a thousand times",
+             ("Hello", "regret"): "I'm sorry|It's no secret",
+             ("One Dance", "desire"): ""})
+
+    def test_codings_reach_the_song(self) -> None:
+        """Test that a stored coding reaches its song through the
+        relationship."""
+        self.__write_codings(self.CODINGS_CSV)
+        self.assertEqual(
+            self.__run_build("--codings", str(self.__codings))[0], 0)
+        session: Session = self.__session()
+        song: Song | None = session.get(Song, 1)
+        assert song is not None
+        self.assertEqual(sorted(x.keyword for x in song.codings),
+                         ["longing", "regret"])
+
+    def test_omitted_codings_leaves_table_empty(self) -> None:
+        """Test that an omitted coding option leaves no codings."""
+        self.__write_codings(self.CODINGS_CSV)
+        status: int
+        stderr: str
+        status, stderr = self.__run_build()
+        self.assertEqual(status, 0)
+        self.assertIn("0 codings", stderr)
+        self.assertEqual(self.__stored_codings(), {})
+
+    def test_codings_unknown_song_fails(self) -> None:
+        """Test that a coding row naming an unknown song fails the
+        build, leaving the previous store contents intact."""
+        self.assertEqual(self.__run_build()[0], 0)
+        titles: dict[int, str] = self.__song_titles()
+        self.__write_codings(
+            "Song,Artist Credit,Keyword,Quote\n"
+            "Nowhere,Nobody,longing,a line\n")
+        status: int
+        stderr: str
+        status, stderr = self.__run_build(
+            "--codings", str(self.__codings))
+        self.assertNotEqual(status, 0)
+        self.assertIn("no song \"Nowhere\" by \"Nobody\"", stderr)
+        self.assertEqual(self.__song_titles(), titles)
+        self.assertEqual(self.__stored_codings(), {})
+
+    def test_codings_duplicated_keyword_fails(self) -> None:
+        """Test that two rows naming the same song and keyword fail
+        the build, without partial data."""
+        self.__write_codings(
+            "Song,Artist Credit,Keyword,Quote\n"
+            "Hello,Adele,longing,a line\n"
+            "Hello,Adele,longing,another line\n")
+        status: int
+        stderr: str
+        status, stderr = self.__run_build(
+            "--codings", str(self.__codings))
+        self.assertNotEqual(status, 0)
+        self.assertIn("duplicated coding", stderr)
+        session: Session = self.__session()
+        self.assertEqual(list(session.scalars(sa.select(Song))), [])
+
+    def test_codings_missing_column_fails(self) -> None:
+        """Test that a coding CSV missing a required column fails
+        the build."""
+        self.__write_codings(
+            "Song,Artist Credit,Keyword\n"
+            "Hello,Adele,longing\n")
+        status: int
+        stderr: str
+        status, stderr = self.__run_build(
+            "--codings", str(self.__codings))
+        self.assertNotEqual(status, 0)
+        self.assertIn("missing column(s): Quote", stderr)
+        self.assertEqual(self.__stored_codings(), {})
+
+    def test_missing_codings_csv_fails(self) -> None:
+        """Test that a given but missing coding CSV fails."""
+        status: int
+        stderr: str
+        status, stderr = self.__run_build(
+            "--codings", str(self.__codings))
+        self.assertNotEqual(status, 0)
+        self.assertIn("error:", stderr)
+        self.assertIn(str(self.__codings), stderr)
+        session: Session = self.__session()
+        self.assertEqual(list(session.scalars(sa.select(Song))), [])
+
+    def test_codings_replaced_on_rebuild(self) -> None:
+        """Test that a rebuild replaces the previous codings rather
+        than adding to them."""
+        self.__write_codings(self.CODINGS_CSV)
+        self.assertEqual(
+            self.__run_build("--codings", str(self.__codings))[0], 0)
+        self.__write_codings(
+            "Song,Artist Credit,Keyword,Quote\n"
+            "Shape of You,Ed Sheeran,attraction,I'm in love with"
+            " your body\n")
+        status: int
+        stderr: str
+        status, stderr = self.__run_build(
+            "--codings", str(self.__codings))
+        self.assertEqual(status, 0)
+        self.assertIn("1 codings", stderr)
+        self.assertEqual(
+            self.__stored_codings(),
+            {("Shape of You", "attraction"):
+                "I'm in love with your body"})

tools/tests/test_models.py

@@ -14,6 +14,7 @@ from pop_fem_audit_tools.database import Base, DataSource
 from pop_fem_audit_tools.models import (
     Artist,
     ChartEntry,
+    Coding,
     Role,
     Song,
     SongArtist,
@@ -47,6 +48,10 @@ class TestModels(unittest.TestCase):
             SongArtist(artist=Artist(name="Wizkid"),
                        role=Role.FEATURED, position=1)]
         song.lyrics = "Baby, I like your style"
+        song.codings = [
+            Coding(keyword="desire",
+                   quotes="Baby, I like your style|One dance"),
+            Coding(keyword="nightlife", quotes="")]
         self.__session.add(song)
         self.__session.commit()
 
@@ -83,6 +88,35 @@ class TestModels(unittest.TestCase):
         with self.assertRaises(sa.exc.IntegrityError):
             self.__session.commit()
 
+    def test_song_codings(self) -> None:
+        """Test reading a song's codings back through the
+        relationship."""
+        self.__add_song()
+        self.__session.expunge_all()
+        song: Song | None = self.__session.scalar(
+            sa.select(Song).where(Song.title == "One Dance"))
+        assert song is not None
+        self.assertEqual(
+            {(x.keyword, x.quotes) for x in song.codings},
+            {("desire", "Baby, I like your style|One dance"),
+             ("nightlife", "")})
+        coding: Coding | None = self.__session.scalar(
+            sa.select(Coding).where(Coding.keyword == "desire"))
+        assert coding is not None
+        self.assertEqual(coding.song.title, "One Dance")
+
+    def test_duplicated_coding_rejected(self) -> None:
+        """Test that a repeated song and keyword fails."""
+        self.__add_song()
+        song: Song | None = self.__session.scalar(
+            sa.select(Song).where(Song.title == "One Dance"))
+        assert song is not None
+        self.__session.add(
+            Coding(song_id=song.id, keyword="desire",
+                   quotes="another line"))
+        with self.assertRaises(sa.exc.IntegrityError):
+            self.__session.commit()
+
     def test_invalid_role_rejected(self) -> None:
         """Test that an invalid song-artist role fails."""
         self.__add_song()