diff --git a/docs/decision-log.md b/docs/decision-log.md index 8884eac..4541d91 100644 --- a/docs/decision-log.md +++ b/docs/decision-log.md @@ -237,3 +237,11 @@ worklist 角色由重跑冪等指令本身承擔；完整性已由 provenance CSV 與 build-db 統計數字佐證；該檔案不承載任何無法重新 產生的資訊。 +- **新增 `export-llm-input` 子命令**：由工作儲存產出 + `run-llm` 的 JSONL 輸入檔（每筆 `id`＋`content`）；此為 + 「歌手背景絕不進 LLM 輸入、歌詞-only」防火牆的執行點 + ——內容只有歌詞，歌曲身分以 `song-` 不透明鍵放 + `custom_id`，不入訊息本體；ID 由 build-db 決定性指派 + 故檔案可再生，依 commit 判準不進 git（且含歌詞全文， + 版權亦不許）；固定匯出全部歌曲，缺歌詞即失敗；單一 + 位置引數收輸出檔路徑，文件範例輸出至 `tools/instance/`。 diff --git a/docs/project-structure.md b/docs/project-structure.md index 4b8d395..287ea61 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -38,6 +38,9 @@ pop-fem-audit/ │ │ │ # from the inputs │ │ ├── config.py # pydantic-settings 設定（.env） │ │ ├── database.py # SQLAlchemy engine / session / Base +│ │ ├── export_llm_input.py # export the LLM input JSONL +│ │ │ # (lyrics only) from the working +│ │ │ # store │ │ ├── fetch_artists.py # fetch artist metadata from │ │ │ # Wikidata into the snapshot CSV │ │ ├── fetch_lyrics.py # fetch missing lyrics from the diff --git a/tools/README.rst b/tools/README.rst index 77f9197..c7d1ae1 100644 --- a/tools/README.rst +++ b/tools/README.rst @@ -49,6 +49,12 @@ fetch-lyrics Fetch lyrics, for ``build-db`` to merge the fetched lyrics into the database. Lyrics are not committed into the repository due to copyright issue. Check ``pop-fem-audit-tools fetch-lyrics -h`` for complete instructions on its usage. +export-llm-input +---------------- + +Export song data for ``run-llm`` to talk to LLM. Check ``pop-fem-audit-tools export-llm-input -h`` for complete instructions on its usage. + + run-llm ------- diff --git a/tools/docs/source/pop_fem_audit_tools.rst b/tools/docs/source/pop_fem_audit_tools.rst index f2b03d9..8462ba7 100644 --- a/tools/docs/source/pop_fem_audit_tools.rst +++ b/tools/docs/source/pop_fem_audit_tools.rst @@ -28,6 +28,14 @@ pop\_fem\_audit\_tools.database module :show-inheritance: :undoc-members: +pop\_fem\_audit\_tools.export\_llm\_input module +------------------------------------------------ + +.. automodule:: pop_fem_audit_tools.export_llm_input + :members: + :show-inheritance: + :undoc-members: + pop\_fem\_audit\_tools.fetch\_artists module -------------------------------------------- diff --git a/tools/src/pop_fem_audit_tools/__main__.py b/tools/src/pop_fem_audit_tools/__main__.py index d728b06..4a1201d 100644 --- a/tools/src/pop_fem_audit_tools/__main__.py +++ b/tools/src/pop_fem_audit_tools/__main__.py @@ -17,6 +17,7 @@ from types import ModuleType from pop_fem_audit_tools import ( build_db, + export_llm_input, fetch_artists, fetch_lyrics, run_llm, @@ -27,6 +28,7 @@ MODULE_PROG: str = "python -m pop_fem_audit_tools" SUBCOMMANDS: dict[str, Callable[[list[str] | None], int]] = { "build-db": build_db.main, + "export-llm-input": export_llm_input.main, "fetch-artists": fetch_artists.main, "fetch-lyrics": fetch_lyrics.main, "run-llm": run_llm.main, diff --git a/tools/src/pop_fem_audit_tools/export_llm_input.py b/tools/src/pop_fem_audit_tools/export_llm_input.py new file mode 100644 index 0000000..524d605 --- /dev/null +++ b/tools/src/pop_fem_audit_tools/export_llm_input.py @@ -0,0 +1,84 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/8/4 +"""The exporter of the LLM input JSONL file. + +Exports the songs from the SQLite working store into the JSONL +input file for the ``run-llm`` subcommand, given as the positional +command-line argument. This is the enforcement point of the +project's lyrics-only firewall: the output carries only the +lyrics text of each song, identified by an opaque song key; no +title, artist, or chart data crosses into the LLM input. +""" +import argparse +import json +import sys +from pathlib import Path + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from .database import ds +from .models import Song + + +def parse_args(argv: list[str] | None) -> argparse.Namespace: + """Parse the command-line arguments. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The parsed arguments. + """ + parser: argparse.ArgumentParser = argparse.ArgumentParser( + description="Export the LLM input JSONL file (lyrics" + " only) from the SQLite working store.") + parser.add_argument( + "output_jsonl", type=Path, + help="the JSONL output file") + return parser.parse_args(argv) + + +def build_lines(session: Session) -> list[str]: + """Build the JSONL lines of every song's lyrics. + + :param session: The database session. + :return: The JSON lines, one per song, ordered by song ID. + :raises ValueError: When a song has no lyrics. + """ + lines: list[str] = [] + song: Song + for song in session.scalars(sa.select(Song).order_by(Song.id)): + if song.lyrics is None: + raise ValueError( + f"song {song.id} \"{song.title}\": no lyrics") + record: dict[str, str] = { + "id": f"song-{song.id}", "content": song.lyrics} + lines.append(json.dumps(record, ensure_ascii=False)) + return lines + + +def main(argv: list[str] | None = None) -> int: + """Export the LLM input JSONL file from the working store. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The exit status: 0 on success, non-zero on failure. + """ + args: argparse.Namespace = parse_args(argv) + session: Session = ds.get_db() + lines: list[str] + try: + lines = build_lines(session) + except (OSError, sa.exc.SQLAlchemyError, ValueError) as error: + print(f"error: {error}", file=sys.stderr) + return 1 + finally: + session.close() + args.output_jsonl.parent.mkdir(parents=True, exist_ok=True) + with open(args.output_jsonl, "w", encoding="utf-8") as file: + line: str + for line in lines: + file.write(line + "\n") + print(f"done: {len(lines)} songs exported", file=sys.stderr) + return 0 diff --git a/tools/tests/test_export_llm_input.py b/tools/tests/test_export_llm_input.py new file mode 100644 index 0000000..6c224af --- /dev/null +++ b/tools/tests/test_export_llm_input.py @@ -0,0 +1,168 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/8/4 +"""Unit tests for the LLM input exporter module.""" +import io +import json +import tempfile +import unittest +from contextlib import redirect_stderr +from pathlib import Path +from typing import Any +from unittest import mock + +from sqlalchemy.orm import Session + +from pop_fem_audit_tools import config, export_llm_input +from pop_fem_audit_tools.database import Base, DataSource +from pop_fem_audit_tools.models import ( + Artist, + Role, + Song, + SongArtist, +) + + +class TestExportLlmInput(unittest.TestCase): + """Test cases for the LLM input exporter.""" + + def setUp(self) -> None: + """Create a temporary working store for the tests.""" + tmp: tempfile.TemporaryDirectory[str] \ + = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.__dir: Path = Path(tmp.name) + self.__output: Path = self.__dir / "llm-input.jsonl" + url: str = f"sqlite:///{self.__dir}/store.sqlite3" + config.set_settings(config.Settings( + SQLALCHEMY_DATABASE_URL=url, + ANTHROPIC_API_KEY="test-key")) + self.__ds: DataSource = DataSource() + patcher: Any = mock.patch.object( + export_llm_input, "ds", self.__ds) + patcher.start() + self.addCleanup(patcher.stop) + + def __seed( + self, songs: list[tuple[str, str, str | None]]) -> None: + """Create the schema and the fixture songs. + + Each song gets a single primary artist at position 0, + whose name equals the song's artist credit; the song IDs + are assigned in list order starting from 1. + + :param songs: The (title, artist, lyrics) triples. + :return: None. + """ + Base.metadata.create_all(self.__ds.engine) + session: Session = self.__ds.get_db() + try: + title: str + artist: str + lyrics: str | None + for title, artist, lyrics in songs: + artist_row: Artist = Artist(name=artist) + song: Song = Song( + title=title, artist_credit=artist, + lyrics=lyrics) + session.add(song) + session.add(SongArtist( + song=song, artist=artist_row, + role=Role.PRIMARY, position=0)) + session.commit() + finally: + session.close() + + def __run_export(self) -> tuple[int, str]: + """Run the exporter with the standard error captured. + + :return: A tuple of the exit status and the standard + error. + """ + stderr: io.StringIO = io.StringIO() + with redirect_stderr(stderr): + status: int = export_llm_input.main( + [str(self.__output)]) + return status, stderr.getvalue() + + @staticmethod + def __read_records(path: Path) -> list[dict[str, str]]: + """Read the JSONL records of a file. + + :param path: The JSONL file. + :return: The parsed records, in file order. + """ + with open(path, encoding="utf-8") as file: + return [json.loads(line) for line in file + if line.strip() != ""] + + def test_exports_songs_ordered_by_id(self) -> None: + """Test that the songs export in ID order, one per line.""" + self.__seed([ + ("Hello", "Adele", "hello lyrics\n"), + ("Umbrella", "Rihanna", "umbrella lyrics\n")]) + status: int + stderr: str + status, stderr = self.__run_export() + self.assertEqual(status, 0) + records: list[dict[str, str]] = self.__read_records( + self.__output) + self.assertEqual(records, [ + {"id": "song-1", "content": "hello lyrics\n"}, + {"id": "song-2", "content": "umbrella lyrics\n"}]) + self.assertIn("done: 2 songs exported", stderr) + + def test_preserves_non_ascii_lyrics(self) -> None: + """Test that non-ASCII lyrics are written verbatim.""" + self.__seed([("Song", "Artist", "非英文歌詞\n")]) + self.assertEqual(self.__run_export()[0], 0) + self.assertEqual( + self.__output.read_text(encoding="utf-8"), + json.dumps( + {"id": "song-1", "content": "非英文歌詞\n"}, + ensure_ascii=False) + "\n") + self.assertIn("非英文歌詞", self.__output.read_text( + encoding="utf-8")) + + def test_no_artist_or_title_data_in_output(self) -> None: + """Test the lyrics-only firewall: no title or artist + leaks into the output.""" + self.__seed([( + "Confidential Title", "Confidential Artist", + "plain lyrics\n")]) + self.assertEqual(self.__run_export()[0], 0) + content: str = self.__output.read_text(encoding="utf-8") + self.assertNotIn("Confidential Title", content) + self.assertNotIn("Confidential Artist", content) + records: list[dict[str, str]] = self.__read_records( + self.__output) + self.assertEqual(records, [ + {"id": "song-1", "content": "plain lyrics\n"}]) + + def test_missing_lyrics_fails(self) -> None: + """Test that a song without lyrics fails without writing + a partial file.""" + self.__seed([ + ("Hello", "Adele", "hello lyrics\n"), + ("Silent", "Nobody", None)]) + status: int + stderr: str + status, stderr = self.__run_export() + self.assertEqual(status, 1) + self.assertIn( + "error: song 2 \"Silent\": no lyrics", stderr) + self.assertFalse(self.__output.exists()) + + def test_creates_parent_directory(self) -> None: + """Test that the output file's parent directory is + created when missing.""" + self.__seed([("Hello", "Adele", "hello lyrics\n")]) + nested: Path = self.__dir / "nested" / "dir" / "out.jsonl" + status: int = export_llm_input.main([str(nested)]) + self.assertEqual(status, 0) + self.assertTrue(nested.exists()) + records: list[dict[str, str]] = self.__read_records( + nested) + self.assertEqual(records, [ + {"id": "song-1", "content": "hello lyrics\n"}])