From 40c41f27349b7d4ea109d96a9e82261568d31335 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BE=9D=E7=91=AA=E8=B2=93?= Date: Fri, 7 Aug 2026 04:28:28 +0800 Subject: [PATCH] Import the settled coding into the working store Co-Authored-By: Claude Opus 5 (1M context) --- docs/project-structure.md | 6 + docs/research-plan.md | 4 +- .../pop_fem_audit_tools/commands/build_db.py | 128 ++++++++++++++- tools/src/pop_fem_audit_tools/models.py | 24 ++- tools/tests/test_build_db.py | 150 ++++++++++++++++++ tools/tests/test_models.py | 34 ++++ 6 files changed, 337 insertions(+), 9 deletions(-) diff --git a/docs/project-structure.md b/docs/project-structure.md index 4756c9b..9346576 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -94,6 +94,12 @@ pop-fem-audit/ - **`prompts/` 檔名不帶版本號**：版本即 git 歷史，失敗的 版本不保留；論文引用的單位是 `runs/` 內隨執行保存的定義檔 快照（每個執行目錄自我完備），不需檔名可指的版本名。 +- **工作儲存的資料表**：`songs`、`chart_entries`、`artists`、 + `song_artists`、`codings`（定案編碼：一歌一標籤一列，`quotes` + 存該標籤所據的歌詞引述，多句以 `|` 相接）。定案表 + `results/codings.csv` 經 `build-db --codings` 匯入，與其餘資料 + 同一交易，儲存不會半建；詳見 `research-plan.md`「資料儲存與 + 模型」。 - **Commit 判準**：能由「committed 輸入＋程式」決定性再生者不 commit（SQLite 工作儲存、LLM 輸入檔）；源頭、捕捉、人工著作 一律以文字 commit。「可再生仍 commit」的例外有二： diff --git a/docs/research-plan.md b/docs/research-plan.md index a80a4cf..434aa38 100644 --- a/docs/research-plan.md +++ b/docs/research-plan.md @@ -64,7 +64,9 @@ - **資料模型**：`songs`（歌曲實體，含 lyrics nullable 欄位）、 `chart_entries`（1 歌—N 筆榜單紀錄）、`artists`（歌手實體： Wikidata QID、性別、型態、曲風、國籍）、`song_artists` - （M—N 關聯：角色、署名順序）。領域不變量（恰 1000 筆榜單、 + （M—N 關聯：角色、署名順序）、`codings`（定案編碼：一歌一 + 標籤一列，`quotes` 為該標籤所據的歌詞引述，多句以 `|` 相接， + 無證據者為空字串）。領域不變量（恰 1000 筆榜單、 每歌至少一 primary 歌手等）檢查內建於 `build-db`，違規即 建置失敗。 - **歌手背景防火牆**：歌手背景資料只進人工解讀階段（證據表、 diff --git a/tools/src/pop_fem_audit_tools/commands/build_db.py b/tools/src/pop_fem_audit_tools/commands/build_db.py index 34ebe5f..e7defcd 100644 --- a/tools/src/pop_fem_audit_tools/commands/build_db.py +++ b/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 diff --git a/tools/src/pop_fem_audit_tools/models.py b/tools/src/pop_fem_audit_tools/models.py index a75ea09..39453a4 100644 --- a/tools/src/pop_fem_audit_tools/models.py +++ b/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.""" diff --git a/tools/tests/test_build_db.py b/tools/tests/test_build_db.py index 506151a..c43e04e 100644 --- a/tools/tests/test_build_db.py +++ b/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"}) diff --git a/tools/tests/test_models.py b/tools/tests/test_models.py index 51a4cd6..dbfd27d 100644 --- a/tools/tests/test_models.py +++ b/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()