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

Separate the clustered keyword list from the keywords to merge

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 04d5095e44
commit 5cf2b8ee8c
5 changed files with 164 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/pop_fem_audit_tools/commands/cluster_keywords.py
+48 -16
View File
@@ -8,11 +8,14 @@ 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 coding keyword set,
given as the third 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.
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.
"""
import argparse
import csv
@@ -49,14 +52,17 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
" sentence embeddings of the pooled"
" keywords.")
parser.add_argument(
"keywords_txt", type=Path,
"pool_txt", type=Path,
help="the pooled keyword list, one keyword per line")
parser.add_argument(
"groups_csv", type=Path,
help="the group membership CSV output file")
parser.add_argument(
"keywords_json", type=Path,
help="the group name keyword JSON output file")
"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")
parser.add_argument(
"--model", default=MODEL,
help=f"the sentence embedding model (default \"{MODEL}\")")
@@ -221,14 +227,35 @@ def write_groups(path: Path, groups: dict[str, list[str]]) -> None:
writer.writerow([group, keyword])
def write_keywords(path: Path,
groups: dict[str, list[str]]) -> None:
def write_keyword_names(path: Path,
groups: dict[str, list[str]]) -> None:
"""Write the group name keyword text file.
Writes a text file holding the lexicographically sorted
group names, one per line, UTF-8, LF line endings, with a
trailing newline. The file records the clustering result
alone; it holds no :data:`EXTRA_KEYWORD` line.
:param path: The path of the keyword text file to write.
:param groups: The keyword members of every group, keyed by
the group's medoid name.
:return: None.
:raises OSError: When the file cannot be written.
"""
names: list[str] = sorted(groups.keys())
path.write_text(
"".join(f"{x}\n" for x in names), encoding="utf-8")
def write_keywords_to_merge(path: Path,
groups: dict[str, list[str]]) -> None:
"""Write the coding keyword set JSON file.
Writes a JSON file holding a single object with one
``keywords`` key, whose value is the lexicographically
sorted list of the group names plus :data:`EXTRA_KEYWORD`,
UTF-8, with a trailing newline.
UTF-8, with a trailing newline. This is the file
``export-llm-input --extras`` consumes.
:param path: The path of the keyword JSON file to write.
:param groups: The keyword members of every group, keyed by
@@ -248,8 +275,10 @@ def main(argv: list[str] | None = None) -> int:
"""Cluster the pooled keywords into the coding groups.
Writes the group membership CSV file, holding the clustering
result alone, and the coding keyword set JSON file, holding
the group names plus :data:`EXTRA_KEYWORD`.
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`.
:param argv: The command-line arguments, or None for
``sys.argv``.
@@ -258,7 +287,7 @@ def main(argv: list[str] | None = None) -> int:
started: float = time.monotonic()
args: argparse.Namespace = parse_args(argv)
try:
keywords: list[str] = load_keywords(args.keywords_txt)
keywords: list[str] = load_keywords(args.pool_txt)
except (OSError, ValueError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
@@ -272,9 +301,12 @@ def main(argv: list[str] | None = None) -> int:
print(f"error: {error}", file=sys.stderr)
return 1
args.groups_csv.parent.mkdir(parents=True, exist_ok=True)
args.keywords_json.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_keywords(args.keywords_json, groups)
write_keyword_names(args.keywords_txt, groups)
write_keywords_to_merge(args.keywords_to_merge_json, groups)
elapsed: str = format_duration(time.monotonic() - started)
print(
f"done: {len(keywords)} keywords clustered into"
tools/tests/test_cluster_keywords.py
+57 -23
View File
@@ -31,9 +31,11 @@ class TestClusterKeywords(unittest.TestCase):
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.__dir: Path = Path(tmp.name)
self.__keywords_txt: Path = self.__dir / "keywords.txt"
self.__pool_txt: Path = self.__dir / "pool.txt"
self.__groups_csv: Path = self.__dir / "groups.csv"
self.__keywords_json: Path = self.__dir / "keywords.json"
self.__keywords_txt: Path = self.__dir / "keywords.txt"
self.__keywords_to_merge_json: Path \
= self.__dir / "keywords-to-merge.json"
@staticmethod
def __two_cluster_vectors() -> Vectors:
@@ -56,13 +58,13 @@ class TestClusterKeywords(unittest.TestCase):
"b-south": (-0.9396926, -0.3420201),
}
def __write_keywords(self, keywords: list[str]) -> None:
def __write_pool(self, keywords: list[str]) -> None:
"""Write the pooled keyword input file.
:param keywords: The keywords, one per line.
:return: None.
"""
self.__keywords_txt.write_text(
self.__pool_txt.write_text(
"".join(f"{x}\n" for x in keywords), encoding="utf-8")
@staticmethod
@@ -95,15 +97,16 @@ class TestClusterKeywords(unittest.TestCase):
standard error.
:param extra_args: Extra command-line arguments appended
after the three positional arguments.
after the four 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.__keywords_txt), str(self.__groups_csv),
str(self.__keywords_json)]
str(self.__pool_txt), str(self.__groups_csv),
str(self.__keywords_txt),
str(self.__keywords_to_merge_json)]
argv.extend(extra_args or [])
fake: Any = self.__fake_encode(
vectors if vectors is not None
@@ -126,19 +129,33 @@ class TestClusterKeywords(unittest.TestCase):
newline="") as file:
return list(csv.reader(file))
def __read_keywords(self) -> list[str]:
"""Read the group name keyword JSON file.
def __read_keyword_names(self) -> list[str]:
"""Read the group name keyword text file.
:return: The group names under the "keywords" key.
:return: The group names, in file order.
"""
text: str = self.__keywords_txt.read_text(
encoding="utf-8")
lines: list[str] = text.split("\n")
if len(lines) > 0 and lines[-1] == "":
lines = lines[:-1]
return lines
def __read_keywords_to_merge(self) -> list[str]:
"""Read the coding keyword set JSON file.
:return: The group names plus :data:`EXTRA_KEYWORD`
under the "keywords" key.
"""
data: dict[str, list[str]] = json.loads(
self.__keywords_json.read_text(encoding="utf-8"))
self.__keywords_to_merge_json.read_text(
encoding="utf-8"))
return data["keywords"]
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_keywords([
self.__write_pool([
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"])
status: int
@@ -161,7 +178,7 @@ class TestClusterKeywords(unittest.TestCase):
keywords: list[str] = [
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"]
self.__write_keywords(keywords)
self.__write_pool(keywords)
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
@@ -169,16 +186,31 @@ class TestClusterKeywords(unittest.TestCase):
self.assertEqual(
sorted(x[1] for x in rows), sorted(keywords))
def test_keywords_json_sorted_medoids(self) -> None:
"""Test that the keyword JSON file holds the sorted medoid
group names plus the extra a-priori keyword."""
self.__write_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"])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
keywords: list[str] = self.__read_keywords()
names: list[str] = self.__read_keyword_names()
self.assertEqual(names, ["a-center", "b-middle"])
self.assertEqual(names, sorted(names))
self.assertNotIn(cluster_keywords.EXTRA_KEYWORD, names)
def test_keywords_to_merge_json_sorted_medoids(self) -> None:
"""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"])
status: int
status, _ = self.__run_cluster()
self.assertEqual(status, 0)
keywords: list[str] = self.__read_keywords_to_merge()
self.assertEqual(
keywords,
["a-center", "b-middle", cluster_keywords.EXTRA_KEYWORD])
@@ -188,7 +220,7 @@ 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_keywords([
self.__write_pool([
"a-left", "a-center", "a-right",
"b-north", "b-middle", "b-south"])
status: int
@@ -201,23 +233,25 @@ class TestClusterKeywords(unittest.TestCase):
def test_duplicate_keyword_rejected(self) -> None:
"""Test that a duplicate keyword line fails the run
without writing any output file."""
self.__write_keywords(["shared", "shared"])
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_json.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.__keywords_txt.write_text("", encoding="utf-8")
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_json.exists())
self.assertFalse(self.__keywords_txt.exists())
self.assertFalse(self.__keywords_to_merge_json.exists())