From 95a05a0631a3cf94c898fcc4d0872479d41a3861 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BE=9D=E7=91=AA=E8=B2=93?= Date: Thu, 6 Aug 2026 19:11:07 +0800 Subject: [PATCH] Add the tally-codings subcommand for the majority vote Co-Authored-By: Claude Opus 5 (1M context) --- docs/decision-log.md | 10 + docs/methodology.md | 20 +- tools/README.rst | 6 + .../source/pop_fem_audit_tools.commands.rst | 8 + tools/src/pop_fem_audit_tools/__main__.py | 2 + .../pop_fem_audit_tools/commands/__init__.py | 23 + .../commands/tally_codings.py | 543 ++++++++++++++++++ tools/tests/test_tally_codings.py | 543 ++++++++++++++++++ 8 files changed, 1146 insertions(+), 9 deletions(-) create mode 100644 tools/src/pop_fem_audit_tools/commands/tally_codings.py create mode 100644 tools/tests/test_tally_codings.py diff --git a/docs/decision-log.md b/docs/decision-log.md index fd87bc3..a991503 100644 --- a/docs/decision-log.md +++ b/docs/decision-log.md @@ -544,3 +544,13 @@ 刪除、`compare-codings` 刪除——其唯一用途是建構仲裁 輸入。已花費的兩次仲裁執行保留於 `run-costs.md` 供總 支出核算。 +- **定案編碼表改為附引述**，推翻同日「只存歌與碼」的 + 決定。當時三個理由裏，「抄一份等於替後續分析先做選擇， + 並抹掉三票是否指向同一句」已由新的表示法解除：`Quote` + 欄收該標籤在計票中各票所給的引述，逐字去重、按 Unicode + 碼位排序、以單一 `|` 相接，相異者全數保留（實測定案的 + 14,665 個碼中，三票引述完全一致者 10,668、相異兩句者 + 3,493、三句者 503）。「庫內重複」一項仍成立，是接受的 + 代價，換得人工判定時證據在手。版權一項先前寫得不準： + 這些引述早已隨三份執行歸檔 commit，寫入定案表不增加 + 任何新的歌詞重製。 diff --git a/docs/methodology.md b/docs/methodology.md index dba5714..708702b 100644 --- a/docs/methodology.md +++ b/docs/methodology.md @@ -102,9 +102,6 @@ - **第三票取全量**：只對前兩次分歧的標籤補問第三票， 計票結果相同；仍採全量執行——全部歌曲、全部關鍵字 ——使三票在同一條件下取得。 -- **定案表只記歌曲與標籤**：引述留在三份執行歸檔，定案 - 表不轉抄。定案即三票的計票結果，每一票各自的引述 - 依據於其歸檔逐筆可查。 - **對邊緣標籤的作用**：兩次執行只分得出「兩次皆標」與 「僅一次標」；三次執行還分得出 3-0 與 2-1，故「定案 編碼中有多少比例僅以一票之差成立」成為可報告的量。 @@ -152,12 +149,17 @@ 檔只規定任務形狀，換詞彙表、換演算法都不必改它。 - **步驟 3 定案**：`tally-codings <執行歸檔 1> <執行歸檔 2> <執行歸檔 3> <輸出 CSV>` 讀三份執行歸檔的 - `output.jsonl`，逐首取標籤鍵集合（引述不參與計票）， - 三份歌曲清單不一致即失敗、不出檔。某首歌的某個標籤 - 於三份中出現至少兩次即寫出一列，產出 - `results/codings.csv`，欄位 `Song`、`Keyword`，一列 - 一個標籤，歌依 ID 數值升序、同一首歌的標籤按字典序， - 換行為 CRLF（同專案其他 CSV）。 + `output.jsonl`，逐首取標籤鍵集合（引述不參與計票）。 + 某首歌的某個標籤於三份中出現至少兩次即寫出一列，產出 + `results/codings.csv`，欄位 `Song`、`Artist Credit`、 + `Keyword`、`Quote`，一列一個標籤。歌名與演出者名銜 + 逐首查工作儲存取得，故本子命令須在 `build-db` 之後 + 執行。`Quote` 收該標籤在計票中的各份執行所給的歌詞 + 引述：各份的引述串接後逐字去重，按 Unicode 碼位排序， + 以單一 `|` 相接（三份執行彼此無先後主從之別，引述之序 + 取決於引述本身）。列序依印出的前三欄依序排：歌名、 + 演出者名銜、標籤，一律以 Unicode 碼位比較，換行為 + CRLF（同專案其他 CSV）。 - **序列化通則**：所有中間檔為 UTF-8，欄序、鍵序與元素 序皆依上列規則明定，無時間戳、無隨機成分；JSON 解析 一律偵測重複鍵，違規即失敗。人讀為主的產物採純文字或 diff --git a/tools/README.rst b/tools/README.rst index d479cf6..8716655 100644 --- a/tools/README.rst +++ b/tools/README.rst @@ -67,6 +67,12 @@ cluster-keywords Deterministically build the coding vocabulary from the two tagging runs' archives, by pooling their keywords per the project's handoff contract and then sentence-embedding and clustering them. Check ``pop-fem-audit-tools cluster-keywords -h`` for complete instructions on its usage. +tally-codings +------------- + +Settle the coding step by a majority of the three coding runs' archives, and write the final coding table, naming every song by its title and stored artist credit from the working store. Check ``pop-fem-audit-tools tally-codings -h`` for complete instructions on its usage. + + Copyright ========= diff --git a/tools/docs/source/pop_fem_audit_tools.commands.rst b/tools/docs/source/pop_fem_audit_tools.commands.rst index 369b03f..19771ff 100644 --- a/tools/docs/source/pop_fem_audit_tools.commands.rst +++ b/tools/docs/source/pop_fem_audit_tools.commands.rst @@ -52,6 +52,14 @@ pop\_fem\_audit\_tools.commands.run\_llm module :show-inheritance: :undoc-members: +pop\_fem\_audit\_tools.commands.tally\_codings module +----------------------------------------------------- + +.. automodule:: pop_fem_audit_tools.commands.tally_codings + :members: + :show-inheritance: + :undoc-members: + Module contents --------------- diff --git a/tools/src/pop_fem_audit_tools/__main__.py b/tools/src/pop_fem_audit_tools/__main__.py index f021ef1..1532753 100644 --- a/tools/src/pop_fem_audit_tools/__main__.py +++ b/tools/src/pop_fem_audit_tools/__main__.py @@ -22,6 +22,7 @@ from .commands import ( fetch_artists_command, fetch_lyrics_command, run_llm_command, + tally_codings_command, ) MODULE_PROG: str = "python -m pop_fem_audit_tools" @@ -34,6 +35,7 @@ SUBCOMMANDS: dict[str, Callable[[list[str] | None], int]] = { "fetch-artists": fetch_artists_command, "fetch-lyrics": fetch_lyrics_command, "run-llm": run_llm_command, + "tally-codings": tally_codings_command, } """The dispatch table from the subcommand name to the tool main.""" diff --git a/tools/src/pop_fem_audit_tools/commands/__init__.py b/tools/src/pop_fem_audit_tools/commands/__init__.py index bdf6d93..adf9455 100644 --- a/tools/src/pop_fem_audit_tools/commands/__init__.py +++ b/tools/src/pop_fem_audit_tools/commands/__init__.py @@ -93,3 +93,26 @@ def run_llm_command(argv: list[str] | None = None) -> int: """ from .run_llm import main return main(argv) + + +def tally_codings_command(argv: list[str] | None = None) -> int: + """Settle the coding by a majority of the three coding runs. + + Writes the final coding table as the given CSV file, holding + the header row ``Song,Artist Credit,Keyword,Quote`` and one + row per keyword at least two of the three runs assign, the + song named by its title and its stored artist credit from the + SQLite working store, and the keyword carrying the pooled, + deduplicated, and sorted lyric quotes of the runs that + assigned it, joined with a single ``|``. Nothing is written + when the three archives do not cover the same songs, a record + is not a successful result, a record's "text" does not parse + to a JSON object of quote string lists, or a song is not in + the working store; the error message names what failed. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The exit status: 0 on success, non-zero on failure. + """ + from .tally_codings import main + return main(argv) diff --git a/tools/src/pop_fem_audit_tools/commands/tally_codings.py b/tools/src/pop_fem_audit_tools/commands/tally_codings.py new file mode 100644 index 0000000..b969089 --- /dev/null +++ b/tools/src/pop_fem_audit_tools/commands/tally_codings.py @@ -0,0 +1,543 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/8/6 +"""The majority tally of the three coding runs. + +Settles the coding step: the same coding definition file is run +three times independently, and this command counts the votes and +writes the final coding table the paper cites, as the CSV file +given as the fourth positional command-line argument. Only the +keyword key sets of the three runs' archived ``output.jsonl`` +files take part in the tally; the lyric quotes never do. A +(song, keyword) pair is written out when at least two of the +three runs assign it, so three votes never tie, and it carries +the lyric quotes of every run that assigned it, pooled, +deduplicated, sorted by Unicode code point, and joined with a +single ``|``: the three runs are peers, so the quote order +follows the text alone. The three +archives must cover exactly the same set of song IDs, every +record must be a successful result, and every record's "text" +must parse to a JSON object; otherwise the tally fails and +nothing is written. + +The archives identify a song as ``song-``, where ```` is +the song's ID in the SQLite working store. The output table does +not carry that ID: every song is looked up in the working store +and written as its title and its stored artist credit instead, so +this command runs after ``build-db``. The step is fully +deterministic; no LLM call is made. +""" +import argparse +import csv +import json +import sys +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, ClassVar + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from ..database import ds +from ..models import Song +from ..utils import format_duration + + +class TallyError(Exception): + """An error that fails the tally.""" + + +@dataclass +class TalliedCodings: + """The codes settled by a majority of the three coding runs.""" + + codings: dict[int, dict[str, str]] + """The joined lyric quotes of every settled keyword of every + song the three runs cover, keyed by the numeric part of the + song ID and then by the keyword, the keywords + lexicographically sorted; a song with no settled keyword maps + to an empty mapping.""" + + @property + def song_count(self) -> int: + """The number of songs the three runs cover. + + :return: The number of songs, those with no settled + keyword included. + """ + return len(self.codings) + + +class CodingTallier: + """The tallier of the three coding runs' keyword votes.""" + + __MAJORITY: int = 2 + """The number of runs that must assign a keyword to a song for + that code to be settled.""" + __MAX_REPORTED_IDS: int = 10 + """The number of song IDs an error message lists before + summarizing the rest as a count.""" + __QUOTE_SEPARATOR: str = "|" + """The separator between the distinct lyric quotes of one + settled code.""" + + def __init__(self, run_dir_1: Path, run_dir_2: Path, + run_dir_3: Path) -> None: + """Set up the tallier of the three coding runs. + + :param run_dir_1: The first run's archive directory, + containing ``output.jsonl``. + :param run_dir_2: The second run's archive directory, + containing ``output.jsonl``. + :param run_dir_3: The third run's archive directory, + containing ``output.jsonl``. + """ + self.__run_dirs: list[Path] = [ + run_dir_1, run_dir_2, run_dir_3] + """The three runs' archive directories, in the given + order.""" + + def run(self) -> TalliedCodings: + """Load the three coding runs and tally their votes. + + Every record of every run must be a successful result + whose "text" parses to a JSON object of keywords mapped to + their lyric quote lists, and the three runs must cover + exactly the same set of song IDs. Only the keyword keys + are counted; the quotes of a settled code are pooled for + the output. Nothing is written. + + :return: The keywords at least two of the three runs + assign, with their joined quotes, of every song the + runs cover. + :raises TallyError: When an ``output.jsonl`` cannot be + read, a line is not a well-formed output record, a + record is not a successful result, a "text" does not + parse to a JSON object of quote string lists, a JSON + document has a duplicate key, a run has two records of + one song, or the three runs do not cover the same + songs. + """ + runs: list[dict[int, dict[str, list[str]]]] + try: + runs = [self.__load_run(x) for x in self.__run_dirs] + self.__check_same_songs(self.__run_dirs, runs) + except (OSError, ValueError) as error: + raise TallyError(str(error)) from error + return TalliedCodings(codings=self.__tally(runs)) + + @classmethod + def __load_run(cls, run_dir: Path) \ + -> dict[int, dict[str, list[str]]]: + """Load and validate the keyword records of one run. + + :param run_dir: The run's archive directory, containing + ``output.jsonl``. + :return: The lyric quotes of every keyword of every song + of the run, keyed by the numeric part of the song ID + and then by the keyword. + :raises OSError: When ``output.jsonl`` cannot be read. + :raises ValueError: When a line is not a well-formed + output record, a record is not a successful result, a + "text" does not parse to a JSON object of quote string + lists, a JSON document has a duplicate key, or the run + has two records of one song. + """ + path: Path = run_dir / "output.jsonl" + text: str = path.read_text(encoding="utf-8") + records: dict[int, dict[str, list[str]]] = {} + line: str + for line in text.split("\n"): + if line.strip() == "": + continue + record: Any = cls.__parse_json(line, str(path)) + if not isinstance(record, dict) or "id" not in record: + raise ValueError( + f"{path}: record without \"id\": {line}") + item_id: Any = record["id"] + if "error" in record or "text" not in record: + raise ValueError( + f"{path}: id {item_id}: not a successful" + " result") + song_id: int = cls.__parse_song_id(item_id, path) + if song_id in records: + raise ValueError( + f"{path}: id {item_id}: duplicate record") + keywords: Any = cls.__parse_json( + record["text"], f"{path}: id {item_id}: \"text\"") + if not isinstance(keywords, dict): + raise ValueError( + f"{path}: id {item_id}: \"text\" does not" + " parse to a JSON object") + records[song_id] = cls.__quote_lists( + keywords, f"{path}: id {item_id}") + return records + + @staticmethod + def __quote_lists(keywords: dict[str, Any], label: str) \ + -> dict[str, list[str]]: + """Validate the lyric quote list of every keyword. + + :param keywords: The parsed "text" object of one record. + :param label: The location of the record, for the error + message. + :return: The lyric quotes of every keyword, in the given + order. + :raises ValueError: When a keyword's value is not a list + of strings. + """ + quotes: dict[str, list[str]] = {} + keyword: str + value: Any + for keyword, value in keywords.items(): + if not isinstance(value, list) \ + or not all(isinstance(x, str) for x in value): + raise ValueError( + f"{label}: keyword \"{keyword}\": the quotes" + " are not a list of strings") + quotes[keyword] = value + return quotes + + @classmethod + def __parse_json(cls, text: str, label: str) -> Any: + """Parse a JSON document, rejecting duplicate keys. + + :param text: The JSON document. + :param label: The location of the document, for the error + message. + :return: The parsed value. + :raises ValueError: When the document is not valid JSON, + or a key appears more than once in one of its + objects. + """ + try: + return json.loads( + text, + object_pairs_hook=cls.__reject_duplicate_keys) + except ValueError as error: + raise ValueError(f"{label}: {error}") from error + + @staticmethod + def __reject_duplicate_keys( + pairs: list[tuple[str, Any]]) -> dict[str, Any]: + """Build a mapping from key-value pairs, rejecting + duplicates. + + :param pairs: The key-value pairs, in document order. + :return: The mapping built from the pairs. + :raises ValueError: When a key appears more than once. + """ + result: dict[str, Any] = {} + key: str + value: Any + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate key \"{key}\"") + result[key] = value + return result + + @staticmethod + def __parse_song_id(item_id: Any, path: Path) -> int: + """Parse the integer song ID out of an item ID. + + :param item_id: The item ID, expected as ``song-``. + :param path: The output file the ID came from, for the + error message. + :return: The parsed song ID. + :raises ValueError: When the item ID is not + ``song-``. + """ + prefix: str = "song-" + if not isinstance(item_id, str) \ + or not item_id.startswith(prefix) \ + or not item_id[len(prefix):].isdigit(): + raise ValueError( + f"{path}: id \"{item_id}\": not in \"song-\"" + " form") + return int(item_id[len(prefix):]) + + @classmethod + def __check_same_songs( + cls, run_dirs: list[Path], + runs: list[dict[int, dict[str, list[str]]]]) -> None: + """Check that the runs cover exactly the same songs. + + :param run_dirs: The runs' archive directories, in the + given order. + :param runs: The runs' records, in the same order. + :return: None. + :raises ValueError: When two runs do not cover the same + set of song IDs. + """ + first: set[int] = set(runs[0]) + index: int + records: dict[int, dict[str, list[str]]] + for index, records in enumerate(runs): + song_ids: set[int] = set(records) + if song_ids == first: + continue + parts: list[str] = [] + missing: list[int] = sorted(first - song_ids) + if len(missing) > 0: + parts.append( + f"missing {cls.__format_ids(missing)}") + extra: list[int] = sorted(song_ids - first) + if len(extra) > 0: + parts.append(f"extra {cls.__format_ids(extra)}") + raise ValueError( + f"{run_dirs[index]} does not cover the same songs" + f" as {run_dirs[0]}: {'; '.join(parts)}") + + @classmethod + def __format_ids(cls, song_ids: list[int]) -> str: + """Format a list of song IDs for an error message. + + :param song_ids: The numeric song IDs, in ascending + order. + :return: The IDs as ``song-``, comma-separated, with + the tail beyond the reporting limit summarized as a + count. + """ + shown: list[int] = song_ids[:cls.__MAX_REPORTED_IDS] + text: str = ", ".join(f"song-{x}" for x in shown) + rest: int = len(song_ids) - len(shown) + if rest > 0: + text = f"{text} and {rest} more" + return text + + @classmethod + def __tally(cls, runs: list[dict[int, dict[str, list[str]]]]) \ + -> dict[int, dict[str, str]]: + """Tally the keyword votes of the runs, song by song. + + The quotes of a settled keyword are those of every run + that assigned it, pooled, deduplicated by exact string, + sorted by Unicode code point, and joined with a single + separator; the three runs are peers, so the order follows + the quotes themselves. + + :param runs: The runs' records, all covering the same set + of song IDs. + :return: The joined quotes of the keywords at least two of + the three runs assign, keyed by the numeric part of + the song ID and then by the keyword, the keywords + lexicographically sorted. + """ + codings: dict[int, dict[str, str]] = {} + song_id: int + for song_id in sorted(runs[0]): + counts: dict[str, int] = {} + quotes: dict[str, list[str]] = {} + records: dict[int, dict[str, list[str]]] + for records in runs: + keyword: str + given: list[str] + for keyword, given in records[song_id].items(): + counts[keyword] = counts.get(keyword, 0) + 1 + quotes.setdefault(keyword, []).extend(given) + codings[song_id] = { + x: cls.__QUOTE_SEPARATOR.join(sorted(set(quotes[x]))) + for x in sorted(counts) + if counts[x] >= cls.__MAJORITY} + return codings + + +@dataclass +class CodingTable: + """The final coding table the paper cites.""" + + RESULT_CODINGS_CSV: ClassVar[str] = "codings.csv" + """The coding table CSV file's conventional name under + ``results/``.""" + __HEADER: ClassVar[tuple[str, str, str, str]] \ + = ("Song", "Artist Credit", "Keyword", "Quote") + """The header row of the coding table CSV file.""" + + rows: list[tuple[str, str, str, str]] + """The data rows, each the song title, the song's stored + artist credit, the settled keyword, and the keyword's joined + lyric quotes, ordered by title, then artist credit, then + keyword, by Unicode code point.""" + + def write(self, output_csv: Path) -> None: + """Write the coding table CSV file. + + Writes an RFC 4180 CSV file, UTF-8, with CRLF line + endings, carrying the header row + ``Song,Artist Credit,Keyword,Quote`` and one row per + settled keyword, in the row order. The parent directory + is created when it does not exist. + + :param output_csv: The output CSV file. + :return: None. + :raises OSError: When the file cannot be written. + """ + output_csv.parent.mkdir(parents=True, exist_ok=True) + with open(output_csv, "w", encoding="utf-8", + newline="") as file: + writer: Any = csv.writer(file) + writer.writerow(self.__HEADER) + writer.writerows(self.rows) + + +class CodingTableBuilder: + """The builder of the final coding table.""" + + def __init__(self, codings: TalliedCodings, + output_csv: Path) -> None: + """Set up the builder of the final coding table. + + :param codings: The settled codes of the three coding + runs. + :param output_csv: The output CSV file that receives the + coding table. + """ + self.__codings: dict[int, dict[str, str]] = codings.codings + """The joined quotes of every settled keyword of every + song, keyed by the numeric part of the song ID and then by + the keyword.""" + self.__output_csv: Path = output_csv + """The output CSV file.""" + + def run(self) -> CodingTable: + """Name the songs from the working store and write the + table. + + Every song of the tally is looked up in the SQLite working + store and written as its title and its stored artist + credit. Writes the coding table CSV file before + returning; nothing is written when the run fails. + + :return: The coding table. + :raises TallyError: When the working store cannot be read, + or a song of the tally is not in it. + :raises OSError: When the output file cannot be written. + """ + table: CodingTable + try: + songs: dict[int, tuple[str, str]] = self.__load_songs() + table = CodingTable( + rows=self.__build_rows(self.__codings, songs)) + except (sa.exc.SQLAlchemyError, ValueError) as error: + raise TallyError(str(error)) from error + table.write(self.__output_csv) + return table + + @staticmethod + def __load_songs() -> dict[int, tuple[str, str]]: + """Load the title and artist credit of every stored song. + + :return: The title and the stored artist credit of every + song, keyed by the song ID. + :raises sqlalchemy.exc.SQLAlchemyError: When the working + store cannot be read. + """ + session: Session = ds.get_db() + try: + song: Song + return { + song.id: (song.title, song.artist_credit) + for song in session.scalars(sa.select(Song))} + finally: + session.close() + + @staticmethod + def __build_rows(codings: dict[int, dict[str, str]], + songs: dict[int, tuple[str, str]]) \ + -> list[tuple[str, str, str, str]]: + """Build the ordered data rows of the coding table. + + :param codings: The joined quotes of every settled keyword + of every song, keyed by the numeric part of the song + ID and then by the keyword. + :param songs: The title and the stored artist credit of + every stored song, keyed by the song ID. + :return: The rows, each the song title, the artist credit, + the keyword, and the keyword's joined quotes, ordered + by title, then artist credit, then keyword, by Unicode + code point. + :raises ValueError: When a song of the tally is not in the + working store. + """ + quotes: dict[tuple[str, str, str], str] = {} + song_id: int + keywords: dict[str, str] + for song_id, keywords in codings.items(): + if song_id not in songs: + raise ValueError( + f"song-{song_id}: not in the working store") + title: str + artist_credit: str + title, artist_credit = songs[song_id] + keyword: str + quote: str + for keyword, quote in keywords.items(): + quotes[(title, artist_credit, keyword)] = quote + key: tuple[str, str, str] + return [(*key, quotes[key]) for key in sorted(quotes)] + + +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="Settle the coding step by a majority of the" + " three coding runs and write the final" + " coding table.") + parser.add_argument( + "run_dir_1", type=Path, + help="the first coding run's archive directory") + parser.add_argument( + "run_dir_2", type=Path, + help="the second coding run's archive directory") + parser.add_argument( + "run_dir_3", type=Path, + help="the third coding run's archive directory") + parser.add_argument( + "output_csv", type=Path, + help="the output CSV file, by convention" + f" results/{CodingTable.RESULT_CODINGS_CSV}") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + """Settle the coding by a majority of the three coding runs. + + Writes the final coding table as the given CSV file, holding + the header row ``Song,Artist Credit,Keyword,Quote`` and one + row per keyword at least two of the three runs assign, the + song named by its title and its stored artist credit from the + SQLite working store, and the keyword carrying the pooled, + deduplicated, and sorted lyric quotes of the runs that + assigned it, joined with a single ``|``. Nothing is written + when the three archives do not cover the same songs, a record + is not a successful result, a record's "text" does not parse + to a JSON object of quote string lists, or a song is not in + the working store; the error message names what failed. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The exit status: 0 on success, non-zero on failure. + """ + started: float = time.monotonic() + args: argparse.Namespace = parse_args(argv) + try: + codings: TalliedCodings = CodingTallier( + args.run_dir_1, args.run_dir_2, args.run_dir_3).run() + table: CodingTable = CodingTableBuilder( + codings, args.output_csv).run() + elapsed: str = format_duration(time.monotonic() - started) + print( + f"Done. Tallied {len(table.rows)} codes across" + f" {codings.song_count} songs. {elapsed} elapsed.", + file=sys.stderr) + except TallyError as error: + print(f"error: {error}", file=sys.stderr) + return 1 + return 0 diff --git a/tools/tests/test_tally_codings.py b/tools/tests/test_tally_codings.py new file mode 100644 index 0000000..27cc2ef --- /dev/null +++ b/tools/tests/test_tally_codings.py @@ -0,0 +1,543 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/8/6 +"""Unit tests for the coding tally module.""" +import csv +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 +from pop_fem_audit_tools.commands import tally_codings +from pop_fem_audit_tools.database import Base, DataSource +from pop_fem_audit_tools.models import Song + + +class TestTallyCodings(unittest.TestCase): + """Test cases for the coding tally.""" + + def setUp(self) -> None: + """Create the run directories and a temporary store.""" + tmp: tempfile.TemporaryDirectory[str] \ + = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.__dir: Path = Path(tmp.name) + self.__runs: list[Path] = [] + number: int + for number in (1, 2, 3): + run_dir: Path = self.__dir / f"run{number}" + run_dir.mkdir() + self.__runs.append(run_dir) + self.__output_csv: Path \ + = self.__dir / "results" / "codings.csv" + 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( + tally_codings, "ds", self.__ds) + patcher.start() + self.addCleanup(patcher.stop) + + def __seed(self, songs: list[tuple[str, str]]) -> None: + """Create the schema and the fixture songs. + + The song IDs are assigned in list order starting from 1. + + :param songs: The (title, artist credit) pairs. + :return: None. + """ + Base.metadata.create_all(self.__ds.engine) + session: Session = self.__ds.get_db() + try: + title: str + artist_credit: str + for title, artist_credit in songs: + session.add(Song( + title=title, artist_credit=artist_credit, + lyrics="la la la")) + session.commit() + finally: + session.close() + + @staticmethod + def __write_output( + run_dir: Path, records: list[dict[str, Any]]) -> None: + """Write the ``output.jsonl`` file of one run. + + :param run_dir: The run's archive directory. + :param records: The envelope records, in file order. + :return: None. + """ + lines: list[str] = [ + json.dumps(x, ensure_ascii=False) for x in records] + (run_dir / "output.jsonl").write_text( + "\n".join(lines) + "\n", encoding="utf-8") + + @staticmethod + def __record(song_id: int, + keywords: dict[str, list[str]]) -> dict[str, Any]: + """Build one successful coding output record. + + :param song_id: The numeric part of the song ID. + :param keywords: The lyric quotes of every assigned + keyword. + :return: The envelope record. + """ + return { + "id": f"song-{song_id}", + "text": json.dumps(keywords, ensure_ascii=False), + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 1}} + + def __write_codings( + self, + runs: list[dict[int, dict[str, list[str]]]]) -> None: + """Write the three runs' output files. + + :param runs: The assigned keywords and their quotes of + every song, per run, in run order. + :return: None. + """ + index: int + songs: dict[int, dict[str, list[str]]] + for index, songs in enumerate(runs): + self.__write_output(self.__runs[index], [ + self.__record(x, songs[x]) for x in songs]) + + def __run_tally(self) -> tuple[int, str]: + """Run the tally over the three run directories. + + :return: A tuple of the exit status and the standard + error. + """ + argv: list[str] = [ + *(str(x) for x in self.__runs), str(self.__output_csv)] + stderr: io.StringIO = io.StringIO() + with redirect_stderr(stderr): + status: int = tally_codings.main(argv) + return status, stderr.getvalue() + + def __read_rows(self) -> list[list[str]]: + """Read the coding table CSV file. + + :return: All rows, including the header row, in file + order. + """ + with open(self.__output_csv, encoding="utf-8", + newline="") as file: + return list(csv.reader(file)) + + def test_majority_of_three_settles_the_code(self) -> None: + """Test that a keyword three or two runs assign is written + out, and one a single run assigns is not.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"all-three": ["q"], "two-of-three": ["q"], + "only-first": ["q"]}}, + {1: {"all-three": ["q"], "two-of-three": ["q"]}}, + {1: {"all-three": ["q"], "only-third": ["q"]}}, + ]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + rows: list[list[str]] = self.__read_rows() + self.assertEqual(rows, [ + ["Song", "Artist Credit", "Keyword", "Quote"], + ["Alpha", "A Singer", "all-three", "q"], + ["Alpha", "A Singer", "two-of-three", "q"], + ]) + + def test_quotes_do_not_take_part_in_the_tally(self) -> None: + """Test that only the keyword keys are tallied, however + the quotes differ between the runs.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"shared": ["one quote"]}}, + {1: {"shared": ["a wholly different quote", "and"]}}, + {1: {"shared": []}}, + ]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "shared", + "a wholly different quote|and|one quote"]]) + + def test_identical_quotes_collapse_to_one(self) -> None: + """Test that the one quote all three runs give is written + once.""" + self.__seed([("Alpha", "A Singer")]) + codings: dict[int, dict[str, list[str]]] \ + = {1: {"kw": ["the same line"]}} + self.__write_codings([codings, codings, codings]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "kw", "the same line"]]) + + def test_distinct_quotes_joined_in_code_point_order(self) -> None: + """Test that the distinct quotes of the runs that assigned + the keyword are joined with a single "|" in Unicode code + point order, whichever run gave which.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["zebra line", "middle line"]}}, + {1: {"kw": ["apple line", "middle line"]}}, + {1: {"kw": ["zebra line"]}}, + ]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "kw", + "apple line|middle line|zebra line"]]) + + def test_quote_with_comma_and_double_quote(self) -> None: + """Test that a quote holding a comma and a double quote is + escaped per RFC 4180 and reads back unchanged.""" + quote: str = "she said \"no\", twice" + self.__seed([("Alpha", "A Singer")]) + codings: dict[int, dict[str, list[str]]] \ + = {1: {"kw": [quote]}} + self.__write_codings([codings, codings, codings]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "kw", quote]]) + self.assertEqual( + self.__output_csv.read_bytes(), + b"Song,Artist Credit,Keyword,Quote\r\n" + b"Alpha,A Singer,kw," + b"\"she said \"\"no\"\", twice\"\r\n") + + def test_empty_quote_lists_yield_an_empty_cell(self) -> None: + """Test that a settled keyword whose runs all gave an empty + quote list carries an empty quote cell.""" + self.__seed([("Alpha", "A Singer")]) + codings: dict[int, dict[str, list[str]]] = {1: {"kw": []}} + self.__write_codings([codings, codings, codings]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "kw", ""]]) + + def test_non_list_quotes_rejected(self) -> None: + """Test that a keyword whose quotes are not a list of + strings fails the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[0], [ + {"id": "song-1", "text": json.dumps({"kw": "q"}), + "stop_reason": "end_turn", "usage": {}}]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("not a list of strings", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_song_named_from_the_working_store(self) -> None: + """Test that the song is written as its title and its + stored artist credit, and the song ID never appears.""" + self.__seed([("Alpha", "A Singer feat. B Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer feat. B Singer", "kw", "q"]]) + text: str = self.__output_csv.read_text(encoding="utf-8") + self.assertNotIn("song-1", text) + + def test_rows_ordered_by_the_printed_columns(self) -> None: + """Test that the rows are ordered by song title, then + artist credit, then keyword, by Unicode code point, not by + the song ID.""" + self.__seed([ + ("Zulu", "Z Singer"), + ("Alpha", "B Singer"), + ("Alpha", "A Singer"), + ]) + codings: dict[int, dict[str, list[str]]] = { + 1: {"beta": ["q"], "alpha": ["q"]}, + 2: {"gamma": ["q"]}, + 3: {"delta": ["q"]}, + } + self.__write_codings([codings, codings, codings]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "delta", "q"], + ["Alpha", "B Singer", "gamma", "q"], + ["Zulu", "Z Singer", "alpha", "q"], + ["Zulu", "Z Singer", "beta", "q"], + ]) + + def test_csv_uses_crlf_line_endings(self) -> None: + """Test that the CSV file uses CRLF line endings and + quotes a value holding a comma.""" + self.__seed([("Alpha, Reprise", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + data: bytes = self.__output_csv.read_bytes() + self.assertEqual( + data, + b"Song,Artist Credit,Keyword,Quote\r\n" + b"\"Alpha, Reprise\",A Singer,kw,q\r\n") + + def test_summary_line(self) -> None: + """Test the closing summary line.""" + self.__seed([("Alpha", "A Singer"), ("Beta", "B Singer")]) + codings: dict[int, dict[str, list[str]]] = { + 1: {"kw": ["q"], "kw2": ["q"]}, 2: {"kw": ["q"]}} + self.__write_codings([codings, codings, codings]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 0) + self.assertIn( + "Done. Tallied 3 codes across 2 songs.", stderr) + + def test_song_without_settled_code_still_counted(self) -> None: + """Test that a song no two runs agree on writes no row but + still counts as a covered song.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"one": ["q"]}}, {1: {"two": ["q"]}}, + {1: {"three": ["q"]}}, + ]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows(), [ + ["Song", "Artist Credit", "Keyword", "Quote"]]) + self.assertIn( + "Done. Tallied 0 codes across 1 songs.", stderr) + + def test_control_character_in_quote_does_not_truncate( + self) -> None: + """Test that a quote holding U+0085, which the generic + line splitting would break the record on, is read whole.""" + self.__seed([("Alpha", "A Singer")]) + quote: str = "a line\u0085another line" + codings: dict[int, dict[str, list[str]]] \ + = {1: {"kw": [quote]}} + self.__write_codings([codings, codings, codings]) + raw: str = (self.__runs[0] / "output.jsonl").read_text( + encoding="utf-8") + self.assertIn("\u0085", raw) + lines: list[str] = raw.split("\n")[:-1] + self.assertEqual(len(lines), 1) + self.assertGreater(len(raw.splitlines()), len(lines)) + status: int + status, _ = self.__run_tally() + self.assertEqual(status, 0) + self.assertEqual(self.__read_rows()[1:], [ + ["Alpha", "A Singer", "kw", quote]]) + + def test_different_song_sets_rejected(self) -> None: + """Test that archives covering different songs fail the + run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer"), ("Beta", "B Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}, 2: {"kw": ["q"]}}, + {1: {"kw": ["q"]}, 2: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("song-2", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_failed_record_rejected(self) -> None: + """Test that a record carrying an "error" field fails the + run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[2], [ + {"id": "song-1", "error": "invalid_request_error"}]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("not a successful result", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_non_json_text_rejected(self) -> None: + """Test that a refusal, whose "text" is not JSON, fails + the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[1], [ + {"id": "song-1", "text": "I cannot help with that.", + "stop_reason": "end_turn", "usage": {}}]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("song-1", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_non_object_text_rejected(self) -> None: + """Test that a "text" JSON value that is not an object + fails the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[0], [ + {"id": "song-1", "text": json.dumps(["kw"]), + "stop_reason": "end_turn", "usage": {}}]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn( + "does not parse to a JSON object", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_duplicate_key_in_text_rejected(self) -> None: + """Test that a "text" JSON object with a duplicate keyword + key fails the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[0], [ + {"id": "song-1", "text": '{"kw": ["a"], "kw": ["b"]}', + "stop_reason": "end_turn", "usage": {}}]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("duplicate key", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_duplicate_song_record_rejected(self) -> None: + """Test that two records of one song in a run fail the run + without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[1], [ + self.__record(1, {"kw": ["q"]}), + self.__record(1, {"kw": ["q"]})]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("duplicate record", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_malformed_item_id_rejected(self) -> None: + """Test that an item ID not in the ``song-`` form + fails the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + self.__write_output(self.__runs[0], [ + {"id": "track-1", "text": json.dumps({"kw": ["q"]}), + "stop_reason": "end_turn", "usage": {}}]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("song-", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_song_missing_from_the_store_rejected(self) -> None: + """Test that a song the working store does not have fails + the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + codings: dict[int, dict[str, list[str]]] = { + 1: {"kw": ["q"]}, 2: {"kw": ["q"]}} + self.__write_codings([codings, codings, codings]) + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("song-2", stderr) + self.assertIn("working store", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_missing_output_file_rejected(self) -> None: + """Test that a run archive without ``output.jsonl`` fails + the run without writing the CSV file.""" + self.__seed([("Alpha", "A Singer")]) + self.__write_codings([ + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + {1: {"kw": ["q"]}}, + ]) + (self.__runs[2] / "output.jsonl").unlink() + status: int + stderr: str + status, stderr = self.__run_tally() + self.assertEqual(status, 1) + self.assertIn("output.jsonl", stderr) + self.assertFalse(self.__output_csv.exists()) + + def test_tallier_writes_nothing(self) -> None: + """Test that the tallier alone settles the codes and + writes no file.""" + self.__write_codings([ + {1: {"kw": ["q"], "solo": ["q"]}}, + {1: {"kw": ["q"]}}, {1: {"kw": ["q"]}}, + ]) + codings: tally_codings.TalliedCodings \ + = tally_codings.CodingTallier(*self.__runs).run() + self.assertEqual(codings.codings, {1: {"kw": "q"}}) + self.assertEqual(codings.song_count, 1) + self.assertFalse(self.__output_csv.exists()) + + def test_builder_failure_raises_tally_error(self) -> None: + """Test that the table builder reports its own failure as + a ``TallyError``, writing no file.""" + self.__seed([("Alpha", "A Singer")]) + codings: tally_codings.TalliedCodings \ + = tally_codings.TalliedCodings( + codings={9: {"kw": "q"}}) + with self.assertRaises(tally_codings.TallyError) as context: + tally_codings.CodingTableBuilder( + codings, self.__output_csv).run() + self.assertIn("song-9", str(context.exception)) + self.assertFalse(self.__output_csv.exists())