imacat/pop-fem-audit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge the keyword pooling into cluster-keywords

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
imacat
2026-08-17 22:38:30 +08:00
co-authored by Claude Opus 5
parent 5cf2b8ee8c
commit de8b8ee678
18 changed files with 526 additions and 626 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/README.rst
+1 -7
View File
@@ -61,16 +61,10 @@ run-llm
A general command that runs specific LLM instructions with the Anthropic API. The API key must be present in the ``.env`` file in the working directory. Check ``pop-fem-audit-tools run-llm -h`` for complete instructions on its usage.
pool-keywords
-------------
Deterministically pool the keywords of the two tagging runs into the clustering step's input, per the project's handoff contract. Check ``pop-fem-audit-tools pool-keywords -h`` for complete instructions on its usage.
cluster-keywords
----------------
Deterministically build the coding vocabulary from the pooled keywords by sentence-embedding and clustering them. Requires the optional ``cluster`` dependency group. Check ``pop-fem-audit-tools cluster-keywords -h`` for complete instructions on its usage.
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. Requires the optional ``cluster`` dependency group. Check ``pop-fem-audit-tools cluster-keywords -h`` for complete instructions on its usage.
Copyright
tools/docs/source/pop_fem_audit_tools.commands.rst
-8
View File
@@ -44,14 +44,6 @@ pop\_fem\_audit\_tools.commands.fetch\_lyrics module
:show-inheritance:
:undoc-members:
pop\_fem\_audit\_tools.commands.pool\_keywords module
-----------------------------------------------------
.. automodule:: pop_fem_audit_tools.commands.pool_keywords
:members:
:show-inheritance:
:undoc-members:
pop\_fem\_audit\_tools.commands.run\_llm module
-----------------------------------------------
tools/src/pop_fem_audit_tools/__main__.py
-2
View File
@@ -21,7 +21,6 @@ from .commands import (
export_llm_input_command,
fetch_artists_command,
fetch_lyrics_command,
pool_keywords_command,
run_llm_command,
)
@@ -34,7 +33,6 @@ SUBCOMMANDS: dict[str, Callable[[list[str] | None], int]] = {
"export-llm-input": export_llm_input_command,
"fetch-artists": fetch_artists_command,
"fetch-lyrics": fetch_lyrics_command,
"pool-keywords": pool_keywords_command,
"run-llm": run_llm_command,
}
"""The dispatch table from the subcommand name to the tool main."""
tools/src/pop_fem_audit_tools/commands/__init__.py
-1
View File
@@ -8,5 +8,4 @@ from .cluster_keywords import main as cluster_keywords_command
from .export_llm_input import main as export_llm_input_command
from .fetch_artists import main as fetch_artists_command
from .fetch_lyrics import main as fetch_lyrics_command
from .pool_keywords import main as pool_keywords_command
from .run_llm import main as run_llm_command
tools/src/pop_fem_audit_tools/commands/cluster_keywords.py
+229 -60
View File
@@ -2,20 +2,29 @@
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/8/5
"""The deterministic clusterer of the pooled keywords.
"""The deterministic vocabulary-building step.
Builds the coding groups from the pooled keyword list, given as
the first positional command-line argument, by sentence-embedding
every keyword and clustering the embeddings: the group membership,
given as the second positional argument, is written as a CSV file
holding the clustering result alone. The group name keywords
alone, given as the third positional argument, are written as a
text file, one per line. The coding keyword set for
``export-llm-input --extras``, given as the fourth positional
argument, is written as a JSON file holding the group name
keywords plus the researcher's a-priori topic term (see
:data:`EXTRA_KEYWORD`). The step is fully deterministic; no LLM
call is made.
Goes from the two tagging runs' archives straight to the coding
vocabulary, writing five fixed-named artifacts under the output
directory given as the third positional command-line argument.
First, the keywords produced by the two runs of the tagging step
are pooled into the pooled keyword list, per the project's handoff
contract: the pool is the plain union of every keyword key observed
across both runs' valid records, exact-string deduplicated and
sorted, written as a plain text file with one keyword per line, as
:data:`SOURCE_KEYWORDS_TXT`. The provenance mapping records where
every keyword came from for audit purposes as a CSV file, as
:data:`SOURCE_PROVENANCE_CSV`; it never enters any LLM input. Then
the coding groups are built from the pooled keyword list by
sentence-embedding every keyword and clustering the embeddings: the
group membership is written as a CSV file holding the clustering
result alone, as :data:`RESULT_GROUPS_CSV`. The group name
keywords alone are written as a text file, one per line, as
:data:`RESULT_KEYWORDS_TXT`. The coding keyword set for
``export-llm-input --extras`` is written as a JSON file holding the
group name keywords plus the researcher's a-priori topic term (see
:data:`EXTRA_KEYWORD`), as :data:`KEYWORDS_TO_MERGE_JSON`. The
step is fully deterministic; no LLM call is made.
"""
import argparse
import csv
@@ -38,6 +47,27 @@ are not installed."""
EXTRA_KEYWORD: str = "women-power"
"""The researcher's a-priori topic term, included in the coding
keyword set although it is not a clustering result."""
SOURCE_KEYWORDS_TXT: str = "source-keywords.txt"
"""The pooled keyword text file's fixed name under the output
directory."""
SOURCE_PROVENANCE_CSV: str = "source-provenance.csv"
"""The keyword provenance CSV file's fixed name under the output
directory."""
RESULT_KEYWORDS_TXT: str = "result-keywords.txt"
"""The group name keyword text file's fixed name under the output
directory."""
RESULT_GROUPS_CSV: str = "groups.csv"
"""The group membership CSV file's fixed name under the output
directory."""
KEYWORDS_TO_MERGE_JSON: str = "keywords-to-merge.json"
"""The coding keyword set JSON file's fixed name under the output
directory."""
type Records = list[tuple[int, dict[str, Any]]]
"""The valid records of one run: (song ID, keyword mapping) pairs."""
type Provenance = dict[str, list[tuple[str, int]]]
"""The occurrences of every keyword, keyed by the keyword."""
def parse_args(argv: list[str] | None) -> argparse.Namespace:
@@ -48,21 +78,22 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
:return: The parsed arguments.
"""
parser: argparse.ArgumentParser = argparse.ArgumentParser(
description="Build the coding groups by clustering the"
" sentence embeddings of the pooled"
" keywords.")
description="Pool the keywords of the two tagging runs and"
" build the coding groups by clustering their"
" sentence embeddings.")
parser.add_argument(
"pool_txt", type=Path,
help="the pooled keyword list, one keyword per line")
"run_dir_1", type=Path,
help="the first tagging run's archive directory")
parser.add_argument(
"groups_csv", type=Path,
help="the group membership CSV output file")
"run_dir_2", type=Path,
help="the second tagging run's archive directory")
parser.add_argument(
"keywords_txt", type=Path,
help="the group name keyword text output file")
parser.add_argument(
"keywords_to_merge_json", type=Path,
help="the coding keyword set JSON output file")
"output_dir", type=Path,
help="the output directory, created if missing, that"
f" receives {SOURCE_KEYWORDS_TXT},"
f" {SOURCE_PROVENANCE_CSV}, {RESULT_KEYWORDS_TXT},"
f" {RESULT_GROUPS_CSV}, and"
f" {KEYWORDS_TO_MERGE_JSON}")
parser.add_argument(
"--model", default=MODEL,
help=f"the sentence embedding model (default \"{MODEL}\")")
@@ -75,30 +106,154 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
return parser.parse_args(argv)
def load_keywords(path: Path) -> list[str]:
"""Load and validate the pooled keyword list.
def reject_duplicate_keys(
pairs: list[tuple[str, Any]]) -> dict[str, Any]:
"""Build a mapping from key-value pairs, rejecting duplicates.
:param path: The path of the pooled keyword text file, one
keyword per line.
:return: The keywords, in file order.
:raises OSError: When the file cannot be read.
:raises ValueError: When the file has no keyword, or a
keyword is duplicated.
: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
def parse_song_id(item_id: str, path: Path) -> int:
"""Parse the integer song ID out of an item ID.
:param item_id: The item ID, expected as ``song-<ID>``.
: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-<ID>``.
"""
prefix: str = "song-"
if not item_id.startswith(prefix) \
or not item_id[len(prefix):].isdigit():
raise ValueError(
f"{path}: id \"{item_id}\": not in \"song-<ID>\" form")
return int(item_id[len(prefix):])
def load_run(run_dir: Path) -> tuple[str, Records]:
"""Load and validate the keyword records of one tagging run.
Records carrying an "error" field are skipped. A "text"
field that fails to parse as JSON is a refusal and is
skipped; a "text" field that parses to anything other than a
JSON object, or whose keys are not unique, fails the run.
:param run_dir: The run's archive directory, containing
``output.jsonl``.
:return: The run label (the directory's basename) and its
valid records, each the song ID and the parsed keyword
mapping, in file order.
:raises OSError: When ``output.jsonl`` cannot be read.
:raises ValueError: When a line is not a well-formed output
record, or a "text" field is invalid per the rules above.
"""
path: Path = run_dir / "output.jsonl"
text: str = path.read_text(encoding="utf-8")
lines: list[str] = text.split("\n")
if len(lines) > 0 and lines[-1] == "":
lines = lines[:-1]
if len(lines) == 0:
raise ValueError(f"{path}: no keywords")
seen: set[str] = set()
keyword: str
for keyword in lines:
if keyword in seen:
records: Records = []
line: str
for line in text.split("\n"):
if line.strip() == "":
continue
record: Any = json.loads(line)
if not isinstance(record, dict) or "id" not in record:
raise ValueError(
f"{path}: duplicate keyword \"{keyword}\"")
seen.add(keyword)
return lines
f"{path}: record without \"id\": {line}")
if "error" in record:
continue
if "text" not in record:
raise ValueError(
f"{path}: id {record['id']}: record without"
" \"text\" or \"error\"")
song_id: int = parse_song_id(record["id"], path)
try:
keywords: Any = json.loads(
record["text"],
object_pairs_hook=reject_duplicate_keys)
except json.JSONDecodeError:
continue
if not isinstance(keywords, dict):
raise ValueError(
f"{path}: id {record['id']}: \"text\" does not"
" parse to a JSON object")
records.append((song_id, keywords))
return run_dir.name, records
def pool_keywords(runs: list[tuple[str, Records]],
) -> tuple[list[str], Provenance]:
"""Pool the keywords of the given tagging runs.
:param runs: The runs, each the run label and its valid
records (song ID, keyword mapping).
:return: The sorted, exact-string-deduplicated keyword list
and the provenance mapping from each keyword to its
occurrences, sorted by (run label, song ID).
"""
provenance: Provenance = {}
label: str
records: Records
for label, records in runs:
song_id: int
keywords: dict[str, Any]
for song_id, keywords in records:
keyword: str
for keyword in keywords:
provenance.setdefault(keyword, []).append(
(label, song_id))
for occurrences in provenance.values():
occurrences.sort()
return sorted(provenance.keys()), provenance
def write_pool(path: Path, keywords: list[str]) -> None:
"""Write the pooled keyword list as the clustering input.
Writes a plain text file, one keyword per line, in the given
order, UTF-8, LF line endings, with a trailing newline.
:param path: The path of the pool text file to write.
:param keywords: The sorted, deduplicated keyword list.
:return: None.
"""
path.write_text(
"".join(f"{keyword}\n" for keyword in keywords),
encoding="utf-8")
def write_provenance(path: Path, provenance: Provenance) -> None:
"""Write the keyword provenance mapping.
Writes a CSV file with the header row
``Keyword,Run,Song``, one row per occurrence, long format.
Rows are sorted by keyword lexicographically, then by run
label, then by song ID.
:param path: The path of the provenance CSV file to write.
:param provenance: The provenance mapping from each keyword
to its occurrences (run label, song ID).
:return: None.
:raises OSError: When the file cannot be written.
"""
keyword: str
with open(path, "w", encoding="utf-8", newline="") as file:
writer: Any = csv.writer(file)
writer.writerow(["Keyword", "Run", "Song"])
for keyword in sorted(provenance.keys()):
label: str
song_id: int
for label, song_id in provenance[keyword]:
writer.writerow([keyword, label, song_id])
def encode_keywords(keywords: list[str], model_name: str,
@@ -272,13 +427,17 @@ def write_keywords_to_merge(path: Path,
def main(argv: list[str] | None = None) -> int:
"""Cluster the pooled keywords into the coding groups.
"""Pool the two tagging runs' keywords and cluster them.
Writes the group membership CSV file, holding the clustering
result alone; the group name keyword text file, holding the
same group names as a readable list; and the coding keyword
set JSON file, holding the group names plus
:data:`EXTRA_KEYWORD`.
Writes the five fixed-named artifacts under the output
directory, creating it (with parents) if it does not exist:
the pooled keyword text file and the keyword provenance CSV
file; then the group membership CSV file, holding the
clustering result alone; the group name keyword text file,
holding the same group names as a readable list; and the
coding keyword set JSON file, holding the group names plus
:data:`EXTRA_KEYWORD`. When the input is rejected, none of
the five files is written.
:param argv: The command-line arguments, or None for
``sys.argv``.
@@ -286,11 +445,17 @@ def main(argv: list[str] | None = None) -> int:
"""
started: float = time.monotonic()
args: argparse.Namespace = parse_args(argv)
run1: tuple[str, Records]
run2: tuple[str, Records]
try:
keywords: list[str] = load_keywords(args.pool_txt)
run1 = load_run(args.run_dir_1)
run2 = load_run(args.run_dir_2)
except (OSError, ValueError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
keywords: list[str]
provenance: Provenance
keywords, provenance = pool_keywords([run1, run2])
try:
embeddings: Any = encode_keywords(
keywords, args.model, args.revision)
@@ -300,16 +465,20 @@ def main(argv: list[str] | None = None) -> int:
except (RuntimeError, ValueError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
args.groups_csv.parent.mkdir(parents=True, exist_ok=True)
args.keywords_txt.parent.mkdir(parents=True, exist_ok=True)
args.keywords_to_merge_json.parent.mkdir(
parents=True, exist_ok=True)
write_groups(args.groups_csv, groups)
write_keyword_names(args.keywords_txt, groups)
write_keywords_to_merge(args.keywords_to_merge_json, groups)
args.output_dir.mkdir(parents=True, exist_ok=True)
write_pool(
args.output_dir / SOURCE_KEYWORDS_TXT, keywords)
write_provenance(
args.output_dir / SOURCE_PROVENANCE_CSV, provenance)
write_groups(args.output_dir / RESULT_GROUPS_CSV, groups)
write_keyword_names(
args.output_dir / RESULT_KEYWORDS_TXT, groups)
write_keywords_to_merge(
args.output_dir / KEYWORDS_TO_MERGE_JSON, groups)
elapsed: str = format_duration(time.monotonic() - started)
print(
f"done: {len(keywords)} keywords clustered into"
f"done: {len(keywords)} keywords pooled from"
f" {len(run1[1])}+{len(run2[1])} records, clustered into"
f" {len(groups)} groups. {elapsed} elapsed.",
file=sys.stderr)
return 0
tools/src/pop_fem_audit_tools/commands/pool_keywords.py
-233
View File
@@ -1,233 +0,0 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/8/5
"""The deterministic pooler of the tagging runs' keywords.
Pools the keywords produced by the two runs of the tagging step
into the clustering step's input, given as the third positional
command-line argument, per the project's handoff contract: the
pool is the plain union of every keyword key observed across both
runs' valid records, exact-string deduplicated and sorted, written
as a plain text file with one keyword per line. The provenance
mapping, given as the fourth positional argument, records where
every keyword came from for audit purposes as a CSV file; it never
enters any LLM input.
"""
import argparse
import csv
import json
import sys
from pathlib import Path
from typing import Any
type Records = list[tuple[int, dict[str, Any]]]
"""The valid records of one run: (song ID, keyword mapping) pairs."""
type Provenance = dict[str, list[tuple[str, int]]]
"""The occurrences of every keyword, keyed by the keyword."""
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="Pool the keywords of the two tagging runs"
" into the clustering step's input.")
parser.add_argument(
"run_dir_1", type=Path,
help="the first tagging run's archive directory")
parser.add_argument(
"run_dir_2", type=Path,
help="the second tagging run's archive directory")
parser.add_argument(
"pool_txt", type=Path,
help="the pooled keyword text output file, one keyword"
" per line")
parser.add_argument(
"provenance_csv", type=Path,
help="the keyword provenance CSV output file")
return parser.parse_args(argv)
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
def parse_song_id(item_id: str, path: Path) -> int:
"""Parse the integer song ID out of an item ID.
:param item_id: The item ID, expected as ``song-<ID>``.
: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-<ID>``.
"""
prefix: str = "song-"
if not item_id.startswith(prefix) \
or not item_id[len(prefix):].isdigit():
raise ValueError(
f"{path}: id \"{item_id}\": not in \"song-<ID>\" form")
return int(item_id[len(prefix):])
def load_run(run_dir: Path) -> tuple[str, Records]:
"""Load and validate the keyword records of one tagging run.
Records carrying an "error" field are skipped. A "text"
field that fails to parse as JSON is a refusal and is
skipped; a "text" field that parses to anything other than a
JSON object, or whose keys are not unique, fails the run.
:param run_dir: The run's archive directory, containing
``output.jsonl``.
:return: The run label (the directory's basename) and its
valid records, each the song ID and the parsed keyword
mapping, in file order.
:raises OSError: When ``output.jsonl`` cannot be read.
:raises ValueError: When a line is not a well-formed output
record, or a "text" field is invalid per the rules above.
"""
path: Path = run_dir / "output.jsonl"
text: str = path.read_text(encoding="utf-8")
records: Records = []
line: str
for line in text.split("\n"):
if line.strip() == "":
continue
record: Any = json.loads(line)
if not isinstance(record, dict) or "id" not in record:
raise ValueError(
f"{path}: record without \"id\": {line}")
if "error" in record:
continue
if "text" not in record:
raise ValueError(
f"{path}: id {record['id']}: record without"
" \"text\" or \"error\"")
song_id: int = parse_song_id(record["id"], path)
try:
keywords: Any = json.loads(
record["text"],
object_pairs_hook=reject_duplicate_keys)
except json.JSONDecodeError:
continue
if not isinstance(keywords, dict):
raise ValueError(
f"{path}: id {record['id']}: \"text\" does not"
" parse to a JSON object")
records.append((song_id, keywords))
return run_dir.name, records
def pool_keywords(runs: list[tuple[str, Records]],
) -> tuple[list[str], Provenance]:
"""Pool the keywords of the given tagging runs.
:param runs: The runs, each the run label and its valid
records (song ID, keyword mapping).
:return: The sorted, exact-string-deduplicated keyword list
and the provenance mapping from each keyword to its
occurrences, sorted by (run label, song ID).
"""
provenance: Provenance = {}
label: str
records: Records
for label, records in runs:
song_id: int
keywords: dict[str, Any]
for song_id, keywords in records:
keyword: str
for keyword in keywords:
provenance.setdefault(keyword, []).append(
(label, song_id))
for occurrences in provenance.values():
occurrences.sort()
return sorted(provenance.keys()), provenance
def write_pool(path: Path, keywords: list[str]) -> None:
"""Write the pooled keyword list as the clustering input.
Writes a plain text file, one keyword per line, in the given
order, UTF-8, LF line endings, with a trailing newline.
:param path: The path of the pool text file to write.
:param keywords: The sorted, deduplicated keyword list.
:return: None.
"""
path.write_text(
"".join(f"{keyword}\n" for keyword in keywords),
encoding="utf-8")
def write_provenance(path: Path, provenance: Provenance) -> None:
"""Write the keyword provenance mapping.
Writes a CSV file with the header row
``Keyword,Run,Song``, one row per occurrence, long format.
Rows are sorted by keyword lexicographically, then by run
label, then by song ID.
:param path: The path of the provenance CSV file to write.
:param provenance: The provenance mapping from each keyword
to its occurrences (run label, song ID).
:return: None.
:raises OSError: When the file cannot be written.
"""
keyword: str
with open(path, "w", encoding="utf-8", newline="") as file:
writer: Any = csv.writer(file)
writer.writerow(["Keyword", "Run", "Song"])
for keyword in sorted(provenance.keys()):
label: str
song_id: int
for label, song_id in provenance[keyword]:
writer.writerow([keyword, label, song_id])
def main(argv: list[str] | None = None) -> int:
"""Pool the two tagging runs' keywords for clustering.
: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)
run1: tuple[str, Records]
run2: tuple[str, Records]
try:
run1 = load_run(args.run_dir_1)
run2 = load_run(args.run_dir_2)
except (OSError, ValueError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
keywords: list[str]
provenance: Provenance
keywords, provenance = pool_keywords([run1, run2])
args.pool_txt.parent.mkdir(parents=True, exist_ok=True)
args.provenance_csv.parent.mkdir(parents=True, exist_ok=True)
write_pool(args.pool_txt, keywords)
write_provenance(args.provenance_csv, provenance)
print(
f"done: {len(keywords)} keywords pooled from"
f" {len(run1[1])}+{len(run2[1])} records", file=sys.stderr)
return 0
tools/tests/test_cluster_keywords.py
+256 -63
View File
@@ -26,16 +26,45 @@ class TestClusterKeywords(unittest.TestCase):
"""Test cases for the keyword clusterer."""
def setUp(self) -> None:
"""Create a temporary directory for the output files."""
"""Create a temporary directory with two run directories."""
tmp: tempfile.TemporaryDirectory[str] \
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.__dir: Path = Path(tmp.name)
self.__pool_txt: Path = self.__dir / "pool.txt"
self.__groups_csv: Path = self.__dir / "groups.csv"
self.__keywords_txt: Path = self.__dir / "keywords.txt"
self.__run1: Path = self.__dir / "run1"
self.__run2: Path = self.__dir / "run2"
self.__run1.mkdir()
self.__run2.mkdir()
self.__output_dir: Path = self.__dir / "output"
self.__source_keywords_txt: Path \
= self.__output_dir \
/ cluster_keywords.SOURCE_KEYWORDS_TXT
self.__source_provenance_csv: Path \
= self.__output_dir \
/ cluster_keywords.SOURCE_PROVENANCE_CSV
self.__result_keywords_txt: Path \
= self.__output_dir \
/ cluster_keywords.RESULT_KEYWORDS_TXT
self.__result_groups_csv: Path \
= self.__output_dir \
/ cluster_keywords.RESULT_GROUPS_CSV
self.__keywords_to_merge_json: Path \
= self.__dir / "keywords-to-merge.json"
= self.__output_dir \
/ cluster_keywords.KEYWORDS_TO_MERGE_JSON
@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 __two_cluster_vectors() -> Vectors:
@@ -58,15 +87,6 @@ class TestClusterKeywords(unittest.TestCase):
"b-south": (-0.9396926, -0.3420201),
}
def __write_pool(self, keywords: list[str]) -> None:
"""Write the pooled keyword input file.
:param keywords: The keywords, one per line.
:return: None.
"""
self.__pool_txt.write_text(
"".join(f"{x}\n" for x in keywords), encoding="utf-8")
@staticmethod
def __fake_encode(vectors: Vectors) -> Any:
"""Build a test double for :func:`encode_keywords`.
@@ -97,16 +117,15 @@ class TestClusterKeywords(unittest.TestCase):
standard error.
:param extra_args: Extra command-line arguments appended
after the four positional arguments.
after the three positional arguments.
:param vectors: The fixed embedding to encode with; the
two-cluster fixture is used when None.
:return: A tuple of the exit status and the standard
error.
"""
argv: list[str] = [
str(self.__pool_txt), str(self.__groups_csv),
str(self.__keywords_txt),
str(self.__keywords_to_merge_json)]
str(self.__run1), str(self.__run2),
str(self.__output_dir)]
argv.extend(extra_args or [])
fake: Any = self.__fake_encode(
vectors if vectors is not None
@@ -119,22 +138,43 @@ class TestClusterKeywords(unittest.TestCase):
argv + ["--clusters", "2"])
return status, stderr.getvalue()
def __read_source_keywords(self) -> list[str]:
"""Read the pooled source keyword text file.
:return: The keyword list, one keyword per line, with the
trailing empty line from the final newline removed.
"""
lines: list[str] = self.__source_keywords_txt.read_text(
encoding="utf-8").split("\n")
self.assertEqual(lines[-1], "")
return lines[:-1]
def __read_source_provenance(self) -> list[list[str]]:
"""Read the source provenance CSV file.
:return: All rows, including the header row, in file
order.
"""
with open(self.__source_provenance_csv, encoding="utf-8",
newline="") as file:
return list(csv.reader(file))
def __read_groups(self) -> list[list[str]]:
"""Read the group membership CSV file.
:return: All rows, including the header row, in file
order.
"""
with open(self.__groups_csv, encoding="utf-8",
with open(self.__result_groups_csv, encoding="utf-8",
newline="") as file:
return list(csv.reader(file))
def __read_keyword_names(self) -> list[str]:
def __read_result_keywords(self) -> list[str]:
"""Read the group name keyword text file.
:return: The group names, in file order.
"""
text: str = self.__keywords_txt.read_text(
text: str = self.__result_keywords_txt.read_text(
encoding="utf-8")
lines: list[str] = text.split("\n")
if len(lines) > 0 and lines[-1] == "":
@@ -152,12 +192,171 @@ class TestClusterKeywords(unittest.TestCase):
encoding="utf-8"))
return data["keywords"]
def test_pools_union_dedup_sorted(self) -> None:
"""Test the union, dedup, and lexicographic ordering, and
the plain one-keyword-per-line source keyword file
shape."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": json.dumps({"a-left": 1, "shared": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-3",
"text": json.dumps({"b-middle": 1, "shared": 1})},
])
vectors: Vectors = {
**self.__two_cluster_vectors(),
"shared": (1.0, 0.0)}
status: int
stderr: str
status, stderr = self.__run_cluster(vectors=vectors)
self.assertEqual(status, 0)
self.assertEqual(
self.__read_source_keywords(),
["a-left", "b-middle", "shared"])
self.assertIn(
"done: 3 keywords pooled from 1+1 records", stderr)
def test_skips_error_records(self) -> None:
"""Test that records carrying an "error" field are
excluded from the pool and the record count."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": json.dumps({"a-left": 1})},
{"id": "song-2", "error": "invalid_request_error"},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"b-middle": 1})},
])
status: int
stderr: str
status, stderr = self.__run_cluster()
self.assertEqual(status, 0)
self.assertEqual(
self.__read_source_keywords(), ["a-left", "b-middle"])
self.assertIn(
"done: 2 keywords pooled from 1+1 records", stderr)
def test_skips_non_json_text_records(self) -> None:
"""Test that a refusal, whose "text" does not parse as
JSON, is skipped rather than failing the run."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": json.dumps({"a-left": 1})},
{"id": "song-2", "text": "I cannot help with that."},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"b-middle": 1})},
])
status: int
stderr: str
status, stderr = self.__run_cluster()
self.assertEqual(status, 0)
self.assertEqual(
self.__read_source_keywords(), ["a-left", "b-middle"])
self.assertIn(
"done: 2 keywords pooled from 1+1 records", stderr)
def test_duplicate_key_in_text_rejected(self) -> None:
"""Test that a "text" JSON object with a duplicate key
fails the run without writing any output file."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": '{"a-left": 1, "a-left": 2}'},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"b-middle": 1})},
])
status: int
stderr: str
status, stderr = self.__run_cluster()
self.assertEqual(status, 1)
self.assertIn("duplicate key", stderr)
self.assertFalse(self.__source_keywords_txt.exists())
self.assertFalse(self.__source_provenance_csv.exists())
self.assertFalse(self.__result_groups_csv.exists())
self.assertFalse(self.__result_keywords_txt.exists())
self.assertFalse(self.__keywords_to_merge_json.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 any output file."""
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(["a-left"])},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"b-middle": 1})},
])
status: int
stderr: str
status, stderr = self.__run_cluster()
self.assertEqual(status, 1)
self.assertIn("song-1", stderr)
self.assertFalse(self.__source_keywords_txt.exists())
self.assertFalse(self.__source_provenance_csv.exists())
self.assertFalse(self.__result_groups_csv.exists())
self.assertFalse(self.__result_keywords_txt.exists())
self.assertFalse(self.__keywords_to_merge_json.exists())
def test_provenance_content_and_ordering(self) -> None:
"""Test the provenance content and its ordering: rows
sorted by keyword lexicographically, then by run label,
then by song ID."""
self.__write_output(self.__run1, [
{"id": "song-2", "text": json.dumps({"shared": 1})},
{"id": "song-1", "text": json.dumps({"shared": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-5",
"text": json.dumps({"shared": 1, "b-middle": 1})},
])
vectors: Vectors = {
**self.__two_cluster_vectors(),
"shared": (1.0, 0.0)}
status: int
status, _ = self.__run_cluster(vectors=vectors)
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_source_provenance()
self.assertEqual(rows[1:], [
["b-middle", "run2", "5"],
["shared", "run1", "1"],
["shared", "run1", "2"],
["shared", "run2", "5"],
])
def test_provenance_file_header_and_row_count(self) -> None:
"""Test that the provenance CSV file starts with the
``Keyword,Run,Song`` header row and has exactly one row
per keyword occurrence."""
self.__write_output(self.__run1, [
{"id": "song-2", "text": json.dumps({"shared": 1})},
{"id": "song-1", "text": json.dumps({"shared": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-5",
"text": json.dumps({"shared": 1, "b-middle": 1})},
])
vectors: Vectors = {
**self.__two_cluster_vectors(),
"shared": (1.0, 0.0)}
status: int
status, _ = self.__run_cluster(vectors=vectors)
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_source_provenance()
self.assertEqual(rows[0], ["Keyword", "Run", "Song"])
self.assertEqual(len(rows), 1 + 4)
def test_groups_csv_header_and_ordering(self) -> None:
"""Test the header row and the group/keyword ordering of
the group membership CSV file."""
self.__write_pool([
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"])
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(
{"a-left": 1, "a-center": 1, "a-right": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-2", "text": json.dumps(
{"b-north": 1, "b-middle": 1, "b-south": 1})},
])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
@@ -178,7 +377,11 @@ class TestClusterKeywords(unittest.TestCase):
keywords: list[str] = [
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"]
self.__write_pool(keywords)
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(
{x: 1 for x in keywords})},
])
self.__write_output(self.__run2, [])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
@@ -187,15 +390,21 @@ class TestClusterKeywords(unittest.TestCase):
sorted(x[1] for x in rows), sorted(keywords))
def test_keywords_txt_sorted_medoids(self) -> None:
"""Test that the keyword text file holds the sorted medoid
group names without the extra a-priori keyword."""
self.__write_pool([
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"])
"""Test that the result keyword text file holds the
sorted medoid group names without the extra a-priori
keyword."""
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(
{"a-left": 1, "a-center": 1, "a-right": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-2", "text": json.dumps(
{"b-north": 1, "b-middle": 1, "b-south": 1})},
])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
names: list[str] = self.__read_keyword_names()
names: list[str] = self.__read_result_keywords()
self.assertEqual(names, ["a-center", "b-middle"])
self.assertEqual(names, sorted(names))
self.assertNotIn(cluster_keywords.EXTRA_KEYWORD, names)
@@ -204,9 +413,14 @@ class TestClusterKeywords(unittest.TestCase):
"""Test that the coding keyword set JSON file holds the
sorted medoid group names plus the extra a-priori
keyword."""
self.__write_pool([
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"])
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(
{"a-left": 1, "a-center": 1, "a-right": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-2", "text": json.dumps(
{"b-north": 1, "b-middle": 1, "b-south": 1})},
])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
@@ -220,38 +434,17 @@ class TestClusterKeywords(unittest.TestCase):
def test_extra_keyword_absent_from_groups_csv(self) -> None:
"""Test that the extra a-priori keyword appears in no row
of the group membership CSV file."""
self.__write_pool([
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"])
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(
{"a-left": 1, "a-center": 1, "a-right": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-2", "text": json.dumps(
{"b-north": 1, "b-middle": 1, "b-south": 1})},
])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_groups()
for row in rows:
self.assertNotIn(cluster_keywords.EXTRA_KEYWORD, row)
def test_duplicate_keyword_rejected(self) -> None:
"""Test that a duplicate keyword line fails the run
without writing any output file."""
self.__write_pool(["shared", "shared"])
status: int
stderr: str
status, stderr = self.__run_cluster()
self.assertEqual(status, 1)
self.assertIn("duplicate keyword", stderr)
self.assertFalse(self.__groups_csv.exists())
self.assertFalse(self.__keywords_txt.exists())
self.assertFalse(self.__keywords_to_merge_json.exists())
def test_empty_input_rejected(self) -> None:
"""Test that an empty keyword file fails the run without
writing any output file."""
self.__pool_txt.write_text("", encoding="utf-8")
status: int
stderr: str
status, stderr = self.__run_cluster()
self.assertEqual(status, 1)
self.assertIn("no keywords", stderr)
self.assertFalse(self.__groups_csv.exists())
self.assertFalse(self.__keywords_txt.exists())
self.assertFalse(self.__keywords_to_merge_json.exists())
tools/tests/test_pool_keywords.py
-217
View File
@@ -1,217 +0,0 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/8/5
"""Unit tests for the keyword pooler 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 pop_fem_audit_tools.commands import pool_keywords
class TestPoolKeywords(unittest.TestCase):
"""Test cases for the keyword pooler."""
def setUp(self) -> None:
"""Create a temporary directory with two run directories."""
tmp: tempfile.TemporaryDirectory[str] \
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.__dir: Path = Path(tmp.name)
self.__run1: Path = self.__dir / "run1"
self.__run2: Path = self.__dir / "run2"
self.__run1.mkdir()
self.__run2.mkdir()
self.__pool: Path = self.__dir / "pool.txt"
self.__provenance: Path = self.__dir / "provenance.csv"
@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")
def __run_pool(self) -> tuple[int, str]:
"""Run the pooler 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 = pool_keywords.main([
str(self.__run1), str(self.__run2),
str(self.__pool), str(self.__provenance)])
return status, stderr.getvalue()
def __read_pool(self) -> list[str]:
"""Read the pool text file.
:return: The keyword list, one keyword per line, with the
trailing empty line from the final newline removed.
"""
lines: list[str] = self.__pool.read_text(
encoding="utf-8").split("\n")
self.assertEqual(lines[-1], "")
return lines[:-1]
def __read_provenance(self) -> list[list[str]]:
"""Read the provenance CSV file.
:return: All rows, including the header row, in file
order.
"""
with open(self.__provenance, encoding="utf-8",
newline="") as file:
return list(csv.reader(file))
def test_pools_union_dedup_sorted(self) -> None:
"""Test the union, dedup, and lexicographic ordering, and
the plain one-keyword-per-line pool file shape."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": json.dumps({"strength": 1, "shared": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-3",
"text": json.dumps({"warrior": 1, "shared": 1})},
])
status: int
stderr: str
status, stderr = self.__run_pool()
self.assertEqual(status, 0)
self.assertEqual(
self.__read_pool(), ["shared", "strength", "warrior"])
self.assertIn(
"done: 3 keywords pooled from 1+1 records", stderr)
def test_skips_error_records(self) -> None:
"""Test that records carrying an "error" field are
excluded from the pool and the record count."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": json.dumps({"strength": 1})},
{"id": "song-2", "error": "invalid_request_error"},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"warrior": 1})},
])
status: int
stderr: str
status, stderr = self.__run_pool()
self.assertEqual(status, 0)
self.assertEqual(
self.__read_pool(), ["strength", "warrior"])
self.assertIn(
"done: 2 keywords pooled from 1+1 records", stderr)
def test_skips_non_json_text_records(self) -> None:
"""Test that a refusal, whose "text" does not parse as
JSON, is skipped rather than failing the run."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": json.dumps({"strength": 1})},
{"id": "song-2", "text": "I cannot help with that."},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"warrior": 1})},
])
status: int
stderr: str
status, stderr = self.__run_pool()
self.assertEqual(status, 0)
self.assertEqual(
self.__read_pool(), ["strength", "warrior"])
self.assertIn(
"done: 2 keywords pooled from 1+1 records", stderr)
def test_duplicate_key_in_text_rejected(self) -> None:
"""Test that a "text" JSON object with a duplicate key
fails the run without writing any output file."""
self.__write_output(self.__run1, [
{"id": "song-1",
"text": '{"strength": 1, "strength": 2}'},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"warrior": 1})},
])
status: int
stderr: str
status, stderr = self.__run_pool()
self.assertEqual(status, 1)
self.assertIn("duplicate key", stderr)
self.assertFalse(self.__pool.exists())
self.assertFalse(self.__provenance.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 any output file."""
self.__write_output(self.__run1, [
{"id": "song-1", "text": json.dumps(["strength"])},
])
self.__write_output(self.__run2, [
{"id": "song-3", "text": json.dumps({"warrior": 1})},
])
status: int
stderr: str
status, stderr = self.__run_pool()
self.assertEqual(status, 1)
self.assertIn("song-1", stderr)
self.assertFalse(self.__pool.exists())
self.assertFalse(self.__provenance.exists())
def test_provenance_content_and_ordering(self) -> None:
"""Test the provenance content and its ordering: rows
sorted by keyword lexicographically, then by run label,
then by song ID."""
self.__write_output(self.__run1, [
{"id": "song-2", "text": json.dumps({"shared": 1})},
{"id": "song-1", "text": json.dumps({"shared": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-5",
"text": json.dumps({"shared": 1, "warrior": 1})},
])
status: int
status, _ = self.__run_pool()
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_provenance()
self.assertEqual(rows[1:], [
["shared", "run1", "1"],
["shared", "run1", "2"],
["shared", "run2", "5"],
["warrior", "run2", "5"],
])
def test_provenance_file_header_and_row_count(self) -> None:
"""Test that the provenance CSV file starts with the
``Keyword,Run,Song`` header row and has exactly one row
per keyword occurrence."""
self.__write_output(self.__run1, [
{"id": "song-2", "text": json.dumps({"shared": 1})},
{"id": "song-1", "text": json.dumps({"shared": 1})},
])
self.__write_output(self.__run2, [
{"id": "song-5",
"text": json.dumps({"shared": 1, "warrior": 1})},
])
status: int
status, _ = self.__run_pool()
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_provenance()
self.assertEqual(rows[0], ["Keyword", "Run", "Song"])
self.assertEqual(len(rows), 1 + 4)