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

Add the tools sub-project with the run_llm batch runner

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
imacat
2026-08-04 14:03:28 +08:00
co-authored by Claude Fable 5
parent 9a42b463a3
commit 12ace6f45a
24 changed files with 1947 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
-2
View File
@@ -1,2 +0,0 @@
# Copy to .env and fill in. .env is gitignored.
ANTHROPIC_API_KEY=sk-ant-...
.gitignore
+1 -6
View File
@@ -7,12 +7,6 @@ data/lyrics/
# Claude Code local settings (machine/user-specific, not research content)
.claude/
# Python
__pycache__/
*.pyc
.venv/
venv/
# IDE
.idea/
.vscode/
@@ -20,3 +14,4 @@ venv/
# OS / Dropbox artifacts
*.conflicted copy*
.DS_Store
excludes/
README.md
+8 -5
View File
@@ -14,17 +14,20 @@ Hot 100（2016–2025）為例的內容分析。
## 重現方式
1. 準備 Python 3.10+ 環境，安裝 `requirements.txt`。
2. 將 Anthropic API 金鑰寫入 `.env`（格式見 `.env.example`）。
3. 依 `docs/research_plan.md` 的階段順序執行 `scripts/` 下的
程式。LLM 步驟使用 `claude-sonnet-4-6`、temperature=0、
1. 準備 Python 3.12+ 環境，安裝分析管線套件：
`pip install -e tools/`。
2. 將 Anthropic API 金鑰寫入 `.env`（格式見
`tools/.env.example`）。
3. 依 `docs/research_plan.md` 的階段順序執行 `tools/`
子專案的程式（如 `python -m pop_fem_audit_tools run-llm ...`）。
LLM 步驟使用 `claude-sonnet-4-6`、temperature=0、
thinking 關閉；每步驟獨立執行兩次後由仲裁步驟合併
（「2+1」協定）。
4. 每次執行的完整紀錄（定義檔快照、原始輸出、參數）存於
`runs/`，可逐筆稽核。論文引用的最終資料表在 `results/`。
注意：歌詞受版權保護，`data/lyrics/` 不隨 repo 發布，須自行
以 `scripts/` 中的抓取程式重建。
以 `tools/` 中的抓取程式重建。
## 授權
docs/decision_log.md
+22
View File
@@ -27,3 +27,25 @@
- **完整歌詞不進 git**（版權）；API 金鑰走 `.env`。
- **專案目錄維持原名 `pop-fem-audit`**：API 管線下目錄名不會
進入模型輸入，無污染疑慮。
## 2026-07-31
- **專案英文名定為「A Feminist Audit of Pop Music」**：明示
女性主義立場，「audit」兼指對歌曲與對 LLM 標籤系統的稽核，
並與縮寫 `pop-fem-audit` 對應。
- **程式碼收整為 `tools/` src-layout 子專案**：發行名
`pop-fem-audit-tools`、import 套件名 `pop_fem_audit_tools`、
description「Tools for A Feminist Audit of Pop Music.」；
以 `pip install -e tools/` 安裝、`python -m
pop_fem_audit_tools.run_llm` 執行；相依套件記於
`pyproject.toml`（`requirements.txt` 移除）；Sphinx 文件
暫緩。理由：後續多支程式將共用程式碼，套件化後測試可用
正常 import；目錄名 `tools/` 經無脈絡的獨立 subagent 命名
評估選出，最誠實反映「服務研究的輔助工具」定位——研究
本體在根目錄的 prompts/、runs/、results/，程式只是配套。
- **CLI 入口改為套件層級 dispatcher**：dispatcher 為單一
入口，`run_llm.py` 的 entry point 移除；有兩種等價呼叫
形式——`python -m pop_fem_audit_tools run-llm ...` 與
console script `pop-fem-audit-tools run-llm ...`
（`[project.scripts]`）。理由：後續多支工具共用單一入口，
`--help` 可列出全部子命令，重現文件穩定。
docs/project_structure.md
+10 -4
View File
@@ -17,10 +17,16 @@ pop-fem-audit/
│ └── lyrics/ # 歌詞快取（gitignored，版權）
├── prompts/ # LLM 定義檔（逐字作為 system prompt）
│ └── <task>_v<N>.md # 版本化：screen_v1.md、judge_v2.md…
├── scripts/ # deterministic Python scripts
│ │ # （runner、抓歌詞、統計、圖表）
│ └── run_llm.py # API runner：2+1 協定、Batch API、
│ # 自動寫入 runs/
├── tools/ # 輔助工具子專案（src-layout）
│ ├── pyproject.toml # 套件 pop_fem_audit_tools；
│ │ # pip install -e tools/ 安裝
│ ├── src/pop_fem_audit_tools/ # deterministic Python 程式
│ │ │ # （runner、抓歌詞、統計、圖表）
│ │ ├── __main__.py # 套件 CLI 進入點（分派子命令）
│ │ └── run_llm.py # API runner：2+1 協定、Batch API、
│ │ # 自動寫入 runs/；執行方式
│ │ # python -m pop_fem_audit_tools run-llm
│ └── tests/ # 單元測試（unittest）
├── runs/ # 每次執行的完整稽核紀錄（進 git）
│ └── <階段>/<日期>-<定義檔版本>/
│ ├── prompt.md # 當次定義檔快照（自我完備）
tools/.env.example
+7
View File
@@ -0,0 +1,7 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
# The Anthropic API key
ANTHROPIC_API_KEY=sk-ant-...
tools/.gitignore
+23
View File
@@ -0,0 +1,23 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
*.pyc
__pycache__
build
dist
*.egg-info
.pytest_cache
venv
.DS_Store
.idea
.claude
.scannerwork
sonar-project.properties
.env
instance
test_temp.py
tools/LICENSE
+202
View File
@@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
tools/MANIFEST.in
+11
View File
@@ -0,0 +1,11 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
include .env.example
recursive-include docs *
recursive-exclude docs/build *
recursive-include tests *
recursive-exclude tests *.pyc
exclude tests/test_temp.py
tools/README.rst
+63
View File
@@ -0,0 +1,63 @@
=======================================
Tools for A Feminist Audit of Pop Music
=======================================
Description
===========
This is a collection of supporting tools for the conference paper "流行音樂中「女性力量」語彙的挪用與污染——以 Billboard Year-End Hot 100（2016–2025）為例的內容分析".
Installation
============
Use ``pip`` to install these tools.
::
% pip install .
This will install the ``pop-fem-audit-tools`` script in the Python environment.
Usage
=====
::
% pop-fem-audit-tools {command} [options] [arguments]
Runs a specific tool command, where "command" can be:
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.
Copyright
=========
Copyright (c) 2026 imacat.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Authors
=======
| imacat
| imacat@mail.imacat.idv.tw
| 2026/7/31
tools/docs/Makefile
+20
View File
@@ -0,0 +1,20 @@
# Minimal makefile for Sphinx documentation
#
# You can set these variables from the command line, and also
# from the environment for the first two.
SPHINXOPTS ?=
SPHINXBUILD ?= sphinx-build
SOURCEDIR = source
BUILDDIR = build
# Put it first so that "make" without argument is like "make help".
help:
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
.PHONY: help Makefile
# Catch-all target: route all unknown targets to Sphinx using the new
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
%: Makefile
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
tools/docs/make.bat
+35
View File
@@ -0,0 +1,35 @@
@ECHO OFF
pushd %~dp0
REM Command file for Sphinx documentation
if "%SPHINXBUILD%" == "" (
set SPHINXBUILD=sphinx-build
)
set SOURCEDIR=source
set BUILDDIR=build
%SPHINXBUILD% >NUL 2>NUL
if errorlevel 9009 (
echo.
echo.The 'sphinx-build' command was not found. Make sure you have Sphinx
echo.installed, then set the SPHINXBUILD environment variable to point
echo.to the full path of the 'sphinx-build' executable. Alternatively you
echo.may add the Sphinx directory to PATH.
echo.
echo.If you don't have Sphinx installed, grab it from
echo.https://www.sphinx-doc.org/
exit /b 1
)
if "%1" == "" goto help
%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
goto end
:help
%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
:end
popd
tools/docs/source/_static/.keep
View File
tools/docs/source/_templates/.keep
View File
tools/docs/source/conf.py
+33
View File
@@ -0,0 +1,33 @@
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
import os
import sys
sys.path.insert(0, os.path.abspath('../../src/'))
import pop_fem_audit_tools
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
project = 'Tools for A Feminist Audit of Pop Music'
copyright = '2026, imacat'
author = 'imacat'
release = pop_fem_audit_tools.VERSION
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
extensions = ["sphinx.ext.autodoc"]
templates_path = ['_templates']
exclude_patterns = []
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
html_theme = 'nature'
html_static_path = ['_static']
tools/docs/source/index.rst
+23
View File
@@ -0,0 +1,23 @@
.. Tools for A Feminist Audit of Pop Music documentation master file, created by
sphinx-quickstart on Fri Jul 31 10:16:41 2026.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
Tools for A Feminist Audit of Pop Music documentation
=====================================================
This is a collection of supporting tools for the conference paper "流行音樂中「女性力量」語彙的挪用與污染——以 Billboard Year-End Hot 100（2016–2025）為例的內容分析".
.. toctree::
:maxdepth: 2
:caption: Contents:
Indices and tables
==================
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`
tools/docs/source/modules.rst
+7
View File
@@ -0,0 +1,7 @@
src
===
.. toctree::
:maxdepth: 4
pop_fem_audit_tools
tools/docs/source/pop_fem_audit_tools.rst
+21
View File
@@ -0,0 +1,21 @@
pop\_fem\_audit\_tools package
==============================
Submodules
----------
pop\_fem\_audit\_tools.run\_llm module
--------------------------------------
.. automodule:: pop_fem_audit_tools.run_llm
:members:
:show-inheritance:
:undoc-members:
Module contents
---------------
.. automodule:: pop_fem_audit_tools
:members:
:show-inheritance:
:undoc-members:
tools/pyproject.toml
+53
View File
@@ -0,0 +1,53 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/30
[project]
name = "pop-fem-audit-tools"
dynamic = ["version"]
description = "Tools for A Feminist Audit of Pop Music."
readme = "README.rst"
requires-python = ">=3.12"
authors = [
{name = "imacat", email = "imacat@mail.imacat.idv.tw"},
]
keywords = [
"content analysis",
"feminism",
"gender studies",
"popular music",
"lyrics",
"Billboard",
"LLM",
"computational social science",
]
classifiers = [
"Development Status :: 3 - Alpha",
"Environment :: Console",
"Intended Audience :: Science/Research",
"License :: OSI Approved :: Apache Software License",
"Natural Language :: English",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Topic :: Scientific/Engineering :: Information Analysis",
"Topic :: Sociology",
"Topic :: Text Processing :: Linguistic",
]
dependencies = [
"anthropic",
]
[project.scripts]
pop-fem-audit-tools = "pop_fem_audit_tools.__main__:main"
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[tool.setuptools.dynamic]
version = {attr = "pop_fem_audit_tools.VERSION"}
tools/src/pop_fem_audit_tools/__init__.py
+10
View File
@@ -0,0 +1,10 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
# AI assistance: Claude Code (Anthropic)
"""Tools for A Feminist Audit of Pop Music."""
VERSION: str = "0.0.0"
"""The package version."""
tools/src/pop_fem_audit_tools/__main__.py
+88
View File
@@ -0,0 +1,88 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
# AI assistance: Claude Code (Anthropic)
"""The package-level command-line entry point.
Dispatches ``python -m pop_fem_audit_tools <subcommand>`` or the
console script ``pop-fem-audit-tools <subcommand>`` to the main
function of the corresponding tool module.
"""
import os.path
import sys
from collections.abc import Callable
from importlib.machinery import ModuleSpec
from types import ModuleType
from pop_fem_audit_tools import run_llm
MODULE_PROG: str = "python -m pop_fem_audit_tools"
"""The program name when run with ``python -m``."""
SUBCOMMANDS: dict[str, Callable[[list[str] | None], int]] = {
"run-llm": run_llm.main,
}
"""The dispatch table from the subcommand name to the tool main."""
def prog() -> str:
"""Return the program name shown in the usage messages.
:return: ``python -m pop_fem_audit_tools`` when run with
``python -m``, or the basename of ``sys.argv[0]`` when run
as a console script.
"""
if sys.argv[0].endswith("__main__.py"):
return MODULE_PROG
return os.path.basename(sys.argv[0])
def usage() -> str:
"""Return the usage text listing the available subcommands.
:return: The usage text.
"""
lines: list[str] = [
f"usage: {prog()} <subcommand> [<argument>...]",
"",
"subcommands:"]
lines.extend(f" {x}" for x in SUBCOMMANDS)
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
"""Dispatch a subcommand to its tool module.
:param argv: The command-line arguments, or None for ``sys.argv``.
:return: The exit status: the status of the subcommand, 0 for the
usage help, or non-zero on a usage error.
"""
args: list[str] = sys.argv[1:] if argv is None else argv
if len(args) == 0:
print("error: missing subcommand", file=sys.stderr)
print(usage(), file=sys.stderr)
return 2
if args[0] in ("-h", "--help"):
print(usage())
return 0
if args[0] not in SUBCOMMANDS:
print(f"error: unknown subcommand \"{args[0]}\"",
file=sys.stderr)
print(usage(), file=sys.stderr)
return 2
prog_backup: str = sys.argv[0]
main_module: ModuleType = sys.modules["__main__"]
spec_backup: ModuleSpec | None = getattr(
main_module, "__spec__", None)
sys.argv[0] = f"{prog()} {args[0]}"
main_module.__spec__ = None
try:
return SUBCOMMANDS[args[0]](args[1:])
finally:
sys.argv[0] = prog_backup
main_module.__spec__ = spec_backup
if __name__ == "__main__":
sys.exit(main())
tools/src/pop_fem_audit_tools/run_llm.py
+623
View File
@@ -0,0 +1,623 @@
#!/usr/bin/env python3
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/30
# AI assistance: Claude Code (Anthropic)
"""The generic batch runner for one LLM analysis step.
Sends every input item to the Anthropic Messages Batch API twice with
the same system prompt, reconciles the disagreeing items with a third
arbitration batch, and archives every artifact self-contained under
``runs/<phase>/<YYYYMMDD-HHMM>-<prompt-stem>/``.
"""
import argparse
import hashlib
import json
import os
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any
import anthropic
type Item = dict[str, str]
"""An input item with "id" and "content"."""
type Result = dict[str, Any]
"""One batch result record."""
type Results = dict[str, Result]
"""The batch result records, keyed by item ID."""
MODEL: str = "claude-sonnet-4-6"
TEMPERATURE: float = 0.0
THINKING: dict[str, str] = {"type": "disabled"}
SCRIPT_VERSION: str = "run_llm.py 1.0.0"
POLL_INTERVAL_SECONDS: float = 60.0
ARBITRATION_TEMPLATE: str = (
"<item>\n{content}\n</item>\n"
"<run1>\n{run1}\n</run1>\n"
"<run2>\n{run2}\n</run2>")
class InputFormatError(Exception):
"""An error in the JSONL input file."""
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="Run one LLM step: 2 runs + 1 arbitration.")
parser.add_argument(
"--prompt", required=True, type=Path,
help="the prompt definition file, used as the system prompt")
parser.add_argument(
"--arbitration-prompt", required=True, type=Path,
help="the arbitration prompt definition file")
parser.add_argument(
"--input", required=True, type=Path,
help="the JSONL input file with \"id\" and \"content\"")
parser.add_argument(
"--phase", required=True,
help="the phase name for the archive directory")
parser.add_argument(
"--max-tokens", type=int, default=2048,
help="the maximum output tokens per request (default 2048)")
parser.add_argument(
"--dry-run", action="store_true",
help="validate and archive without calling the API")
return parser.parse_args(argv)
def validate_item(data: Any, path: Path, number: int) -> Item:
"""Validate one parsed JSONL record as an input item.
:param data: The parsed JSON value of the line.
:param path: The path of the JSONL input file, for the messages.
:param number: The line number, for the messages.
:return: The validated item with "id" and "content".
:raises InputFormatError: When the record is malformed.
"""
if not isinstance(data, dict):
raise InputFormatError(
f"{path}: line {number}: not a JSON object")
if set(data.keys()) != {"id", "content"}:
raise InputFormatError(
f"{path}: line {number}: keys must be exactly"
" \"id\" and \"content\"")
if not isinstance(data["id"], str) or data["id"] == "":
raise InputFormatError(
f"{path}: line {number}: \"id\" must be a"
" non-empty string")
if not isinstance(data["content"], str):
raise InputFormatError(
f"{path}: line {number}: \"content\" must be a"
" string")
return {"id": data["id"], "content": data["content"]}
def load_items(path: Path) -> list[Item]:
"""Load and validate the JSONL input items.
:param path: The path of the JSONL input file.
:return: The items, each with "id" and "content", in file order.
:raises InputFormatError: When a line is malformed, an ID is
duplicated, or the file contains no item.
:raises OSError: When the file cannot be read.
"""
items: list[Item] = []
seen: set[str] = set()
with open(path, encoding="utf-8") as file:
for number, line in enumerate(file, start=1):
if line.strip() == "":
continue
try:
data: Any = json.loads(line)
except json.JSONDecodeError as error:
raise InputFormatError(
f"{path}: line {number}: malformed JSON: {error}")
item: Item = validate_item(data, path, number)
if item["id"] in seen:
raise InputFormatError(
f"{path}: line {number}: duplicated ID"
f" \"{item['id']}\"")
seen.add(item["id"])
items.append(item)
if len(items) == 0:
raise InputFormatError(f"{path}: no input items")
return items
def parse_env_file(path: Path) -> dict[str, str]:
"""Parse a simple KEY=VALUE .env file.
Blank lines and lines starting with "#" are ignored.
:param path: The path of the .env file.
:return: The key-value pairs; empty when the file is missing.
"""
values: dict[str, str] = {}
if not path.is_file():
return values
for line in path.read_text(encoding="utf-8").splitlines():
stripped: str = line.strip()
if stripped == "" or stripped.startswith("#"):
continue
if "=" not in stripped:
continue
key, _, value = stripped.partition("=")
values[key.strip()] = value.strip()
return values
def resolve_api_key(env_path: Path) -> str:
"""Resolve the Anthropic API key.
The ``ANTHROPIC_API_KEY`` environment variable takes precedence;
the .env file is consulted as a fallback.
:param env_path: The path of the .env file.
:return: The API key.
:raises RuntimeError: When no API key can be found.
"""
key: str | None = os.environ.get("ANTHROPIC_API_KEY")
if key:
return key
key = parse_env_file(env_path).get("ANTHROPIC_API_KEY")
if key:
return key
raise RuntimeError(
"ANTHROPIC_API_KEY is not set in the environment and not"
f" found in {env_path}")
def build_request(item: Item, system_prompt: str,
max_tokens: int) -> dict[str, Any]:
"""Build one Message Batches request for an input item.
:param item: The input item with "id" and "content".
:param system_prompt: The system prompt text.
:param max_tokens: The maximum output tokens.
:return: The batch request with "custom_id" and "params".
"""
return {
"custom_id": item["id"],
"params": {
"model": MODEL,
"max_tokens": max_tokens,
"temperature": TEMPERATURE,
"thinking": THINKING,
"system": system_prompt,
"messages": [
{"role": "user", "content": item["content"]},
],
},
}
def build_arbitration_content(content: str, run1_text: str,
run2_text: str) -> str:
"""Build the arbitration user message for one item.
:param content: The original item content.
:param run1_text: The run-1 output text.
:param run2_text: The run-2 output text.
:return: The user message text.
"""
return ARBITRATION_TEMPLATE.format(
content=content, run1=run1_text, run2=run2_text)
def submit_batch(client: anthropic.Anthropic,
requests: list[dict[str, Any]]) -> str:
"""Submit one message batch.
:param client: The Anthropic client.
:param requests: The batch requests.
:return: The batch ID.
"""
return client.messages.batches.create(requests=requests).id
def poll_batches(client: anthropic.Anthropic,
batch_ids: list[str]) -> dict[str, Any]:
"""Poll the batches until every one of them has ended.
Progress is printed to the standard error every poll.
:param client: The Anthropic client.
:param batch_ids: The batch IDs to poll.
:return: The final batch object of each batch, keyed by batch ID.
"""
while True:
batches: dict[str, Any] = {
x: client.messages.batches.retrieve(x) for x in batch_ids}
pending: list[str] = [
x for x in batch_ids
if batches[x].processing_status != "ended"]
for batch_id in batch_ids:
status: str = batches[batch_id].processing_status
print(f"batch {batch_id}: {status}", file=sys.stderr)
if len(pending) == 0:
return batches
time.sleep(POLL_INTERVAL_SECONDS)
def usage_to_dict(usage: Any) -> dict[str, Any]:
"""Convert a usage object to a plain dictionary.
:param usage: The usage object of a message.
:return: The usage as a dictionary, without null entries.
"""
if hasattr(usage, "model_dump"):
return {k: v for k, v in usage.model_dump().items()
if v is not None}
return dict(usage)
def collect_results(client: anthropic.Anthropic,
batch_id: str) -> Results:
"""Collect the results of an ended batch.
A succeeded result carries "text", "stop_reason", and "usage";
any other result carries "error" instead.
:param client: The Anthropic client.
:param batch_id: The batch ID.
:return: The result records, keyed by custom ID.
"""
results: Results = {}
for entry in client.messages.batches.results(batch_id):
result: Any = entry.result
record: Result
match result.type:
case "succeeded":
message: Any = result.message
text: str = "".join(
x.text for x in message.content
if x.type == "text")
record = {"id": entry.custom_id, "text": text,
"stop_reason": message.stop_reason,
"usage": usage_to_dict(message.usage)}
case "errored":
error_type: Any = getattr(
result.error, "type", "unknown")
record = {"id": entry.custom_id,
"error": str(error_type)}
case other:
record = {"id": entry.custom_id,
"error": str(other)}
results[entry.custom_id] = record
return results
def find_failures(item_ids: list[str],
results: Results) -> list[str]:
"""Find the item IDs that failed in a result set.
An item failed when it is missing from the results or when its
record carries an "error" field.
:param item_ids: The item IDs to check, in order.
:param results: The result records, keyed by item ID.
:return: The failed item IDs, in the given order.
"""
return [x for x in item_ids
if x not in results or "error" in results[x]]
def split_by_agreement(
items: list[Item], run1: Results, run2: Results,
) -> tuple[list[str], list[str]]:
"""Split the item IDs into agreed and disagreeing ones.
Two outputs agree when their texts are identical after strip().
:param items: The input items.
:param run1: The run-1 results, keyed by item ID.
:param run2: The run-2 results, keyed by item ID.
:return: A tuple of the agreed item IDs and the disagreeing item
IDs, both in input order.
"""
agreed: list[str] = []
disagreed: list[str] = []
for item in items:
item_id: str = item["id"]
text1: str = run1[item_id]["text"].strip()
text2: str = run2[item_id]["text"].strip()
if text1 == text2:
agreed.append(item_id)
else:
disagreed.append(item_id)
return agreed, disagreed
def build_final_records(
items: list[Item], run1: Results, arbitration: Results,
) -> list[dict[str, str]]:
"""Assemble the final records, one per item, in input order.
An arbitrated item takes the arbitration output; an agreed item
takes the agreed (stripped) run text.
:param items: The input items.
:param run1: The run-1 results, keyed by item ID.
:param arbitration: The arbitration results, keyed by item ID.
:return: The final records with "id", "text", and "source".
"""
records: list[dict[str, str]] = []
for item in items:
item_id: str = item["id"]
if item_id in arbitration:
records.append({"id": item_id,
"text": arbitration[item_id]["text"],
"source": "arbitration"})
else:
records.append({"id": item_id,
"text": run1[item_id]["text"].strip(),
"source": "agreed"})
return records
def create_archive_dir(runs_root: Path, phase: str, prompt_path: Path,
now: datetime) -> Path:
"""Create the archive directory for this execution.
:param runs_root: The root directory of the run archives.
:param phase: The phase name.
:param prompt_path: The path of the prompt definition file.
:param now: The local timestamp of this execution.
:return: The created archive directory.
:raises FileExistsError: When the directory already exists.
"""
stem: str = prompt_path.name
if stem.endswith(".md"):
stem = stem[:-len(".md")]
name: str = f"{now.strftime('%Y%m%d-%H%M')}-{stem}"
directory: Path = runs_root / phase / name
if directory.exists():
raise FileExistsError(
f"archive directory {directory} already exists")
directory.mkdir(parents=True)
return directory
def write_jsonl(path: Path, records: list[dict[str, Any]]) -> None:
"""Write records to a file as JSON Lines.
:param path: The path of the file to write.
:param records: The records, one per line.
"""
with open(path, "w", encoding="utf-8") as file:
for record in records:
file.write(json.dumps(record, ensure_ascii=False) + "\n")
def write_json(path: Path, data: dict[str, Any]) -> None:
"""Write data to a file as pretty-printed JSON.
:param path: The path of the file to write.
:param data: The data to write.
"""
path.write_text(
json.dumps(data, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8")
def sha256_of(path: Path) -> str:
"""Calculate the SHA-256 digest of a file.
:param path: The path of the file.
:return: The hexadecimal SHA-256 digest.
"""
with open(path, "rb") as file:
return hashlib.file_digest(file, "sha256").hexdigest()
def now_iso() -> str:
"""Return the current local time in ISO 8601 format.
:return: The current local time with the timezone offset.
"""
return datetime.now().astimezone().isoformat(timespec="seconds")
def format_timestamp(value: Any) -> str:
"""Format a timestamp value from a batch object as a string.
:param value: The timestamp: a datetime, a string, or None.
:return: The timestamp as a string, or the current local time
when the value is missing.
"""
match value:
case datetime():
return value.isoformat()
case str():
return value
case _:
return now_iso()
def execute_runs(
client: anthropic.Anthropic, items: list[Item],
system_prompt: str, max_tokens: int, meta: dict[str, Any],
) -> tuple[Results, Results]:
"""Submit the two identical runs and await their results.
The batch IDs and timestamps are recorded into the metadata as an
observable side effect.
:param client: The Anthropic client.
:param items: The input items.
:param system_prompt: The system prompt text.
:param max_tokens: The maximum output tokens per request.
:param meta: The metadata to record the batch bookkeeping into.
:return: The results of run 1 and run 2, keyed by item ID.
"""
requests: list[dict[str, Any]] = [
build_request(x, system_prompt, max_tokens) for x in items]
batch_ids: dict[str, str] = {}
for run_name in ("run1", "run2"):
batch_id: str = submit_batch(client, requests)
batch_ids[run_name] = batch_id
meta["batches"][run_name] = {
"batch_id": batch_id, "submitted_at": now_iso(),
"ended_at": None}
print(f"{run_name}: submitted batch {batch_id}",
file=sys.stderr)
batches: dict[str, Any] = poll_batches(
client, list(batch_ids.values()))
for run_name, batch_id in batch_ids.items():
meta["batches"][run_name]["ended_at"] = format_timestamp(
getattr(batches[batch_id], "ended_at", None))
return (collect_results(client, batch_ids["run1"]),
collect_results(client, batch_ids["run2"]))
def execute_arbitration(
client: anthropic.Anthropic, items: list[Item],
disagreed: list[str], run1: Results, run2: Results,
system_prompt: str, max_tokens: int, meta: dict[str, Any],
) -> Results:
"""Submit the arbitration batch and await its results.
The batch ID and timestamps are recorded into the metadata as an
observable side effect.
:param client: The Anthropic client.
:param items: The input items.
:param disagreed: The disagreeing item IDs.
:param run1: The run-1 results, keyed by item ID.
:param run2: The run-2 results, keyed by item ID.
:param system_prompt: The arbitration system prompt text.
:param max_tokens: The maximum output tokens per request.
:param meta: The metadata to record the batch bookkeeping into.
:return: The arbitration results, keyed by item ID.
"""
content_by_id: dict[str, str] = {
x["id"]: x["content"] for x in items}
requests: list[dict[str, Any]] = [
build_request(
{"id": x,
"content": build_arbitration_content(
content_by_id[x], run1[x]["text"], run2[x]["text"])},
system_prompt, max_tokens)
for x in disagreed]
batch_id: str = submit_batch(client, requests)
meta["batches"]["arbitration"] = {
"batch_id": batch_id, "submitted_at": now_iso(),
"ended_at": None}
print(f"arbitration: submitted batch {batch_id}", file=sys.stderr)
batches: dict[str, Any] = poll_batches(client, [batch_id])
meta["batches"]["arbitration"]["ended_at"] = format_timestamp(
getattr(batches[batch_id], "ended_at", None))
return collect_results(client, batch_id)
def main(argv: list[str] | None = None) -> int:
"""Run one LLM step end-to-end.
: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)
try:
items: list[Item] = load_items(args.input)
prompt_text: str = args.prompt.read_text(encoding="utf-8")
arbitration_text: str = args.arbitration_prompt.read_text(
encoding="utf-8")
except (OSError, InputFormatError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
try:
run_dir: Path = create_archive_dir(
Path("runs"), args.phase, args.prompt, datetime.now())
except FileExistsError as error:
print(f"error: {error}", file=sys.stderr)
return 1
meta_path: Path = run_dir / "meta.json"
(run_dir / "prompt.md").write_bytes(args.prompt.read_bytes())
(run_dir / "arbitration_prompt.md").write_bytes(
args.arbitration_prompt.read_bytes())
meta: dict[str, Any] = {
"model": MODEL,
"temperature": TEMPERATURE,
"max_tokens": args.max_tokens,
"thinking": THINKING,
"prompt_path": str(args.prompt),
"prompt_sha256": sha256_of(args.prompt),
"arbitration_prompt_path": str(args.arbitration_prompt),
"arbitration_prompt_sha256": sha256_of(
args.arbitration_prompt),
"batches": {},
"item_count": len(items),
"agreed_count": None,
"agreement_rate": None,
"dry_run": args.dry_run,
"script_version": SCRIPT_VERSION,
}
if args.dry_run:
write_json(meta_path, meta)
print(json.dumps(
build_request(items[0], prompt_text, args.max_tokens),
ensure_ascii=False, indent=2))
print(f"dry run: archive created at {run_dir}",
file=sys.stderr)
return 0
try:
api_key: str = resolve_api_key(Path(".env"))
except RuntimeError as error:
print(f"error: {error}", file=sys.stderr)
return 1
client: anthropic.Anthropic = anthropic.Anthropic(
api_key=api_key)
run1: Results
run2: Results
run1, run2 = execute_runs(
client, items, prompt_text, args.max_tokens, meta)
item_ids: list[str] = [x["id"] for x in items]
write_jsonl(run_dir / "run1.jsonl",
[run1[x] for x in item_ids if x in run1])
write_jsonl(run_dir / "run2.jsonl",
[run2[x] for x in item_ids if x in run2])
failed: set[str] = (set(find_failures(item_ids, run1))
| set(find_failures(item_ids, run2)))
if len(failed) > 0:
write_json(meta_path, meta)
names: str = ", ".join(x for x in item_ids if x in failed)
print(f"error: failed items: {names}", file=sys.stderr)
return 1
agreed: list[str]
disagreed: list[str]
agreed, disagreed = split_by_agreement(items, run1, run2)
meta["agreed_count"] = len(agreed)
meta["agreement_rate"] = len(agreed) / len(items)
arbitration: Results = {}
if len(disagreed) > 0:
arbitration = execute_arbitration(
client, items, disagreed, run1, run2, arbitration_text,
args.max_tokens, meta)
write_jsonl(run_dir / "arbitration.jsonl",
[arbitration[x] for x in disagreed
if x in arbitration])
arb_failed: list[str] = find_failures(disagreed, arbitration)
if len(arb_failed) > 0:
write_json(meta_path, meta)
names = ", ".join(arb_failed)
print(f"error: failed arbitration items: {names}",
file=sys.stderr)
return 1
write_jsonl(run_dir / "final.jsonl",
build_final_records(items, run1, arbitration))
write_json(meta_path, meta)
print(f"done: {len(items)} items, {len(agreed)} agreed,"
f" {len(disagreed)} arbitrated; archived to {run_dir}",
file=sys.stderr)
return 0
tools/tests/test_main.py
+130
View File
@@ -0,0 +1,130 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
# AI assistance: Claude Code (Anthropic)
"""Unit tests for the package-level CLI dispatcher."""
import argparse
import io
import sys
import unittest
from contextlib import redirect_stderr, redirect_stdout
from unittest import mock
from pop_fem_audit_tools import __main__, run_llm
class TestDispatcher(unittest.TestCase):
"""Test cases for the package-level CLI dispatcher."""
@staticmethod
def __run_main(argv: list[str]) -> tuple[int, str, str]:
"""Run the dispatcher with captured output.
:param argv: The command-line arguments.
:return: A tuple of the exit status, the standard output,
and the standard error.
"""
stdout: io.StringIO = io.StringIO()
stderr: io.StringIO = io.StringIO()
with redirect_stdout(stdout), redirect_stderr(stderr):
status: int = __main__.main(argv)
return status, stdout.getvalue(), stderr.getvalue()
def test_run_llm_registered(self) -> None:
"""Test that run-llm is bound to the run_llm tool main."""
self.assertIs(__main__.SUBCOMMANDS["run-llm"], run_llm.main)
def test_run_llm_dispatch(self) -> None:
"""Test that run-llm forwards the arguments and the status."""
tool: mock.Mock = mock.Mock(return_value=7)
with mock.patch.dict(__main__.SUBCOMMANDS,
{"run-llm": tool}):
status: int = self.__run_main(
["run-llm", "--input", "items.jsonl"])[0]
self.assertEqual(status, 7)
tool.assert_called_once_with(["--input", "items.jsonl"])
def test_help_exits_zero_and_lists_subcommands(self) -> None:
"""Test that --help lists the subcommands and exits 0."""
status: int
stdout: str
stderr: str
status, stdout, stderr = self.__run_main(["--help"])
self.assertEqual(status, 0)
self.assertIn("run-llm", stdout)
def test_no_arguments_exits_non_zero(self) -> None:
"""Test that no arguments yields the usage and non-zero."""
status: int
stdout: str
stderr: str
status, stdout, stderr = self.__run_main([])
self.assertNotEqual(status, 0)
self.assertIn("run-llm", stderr)
def test_usage_prog_module_run(self) -> None:
"""Test the usage program name when run with python -m."""
argv: list[str] = ["/x/pop_fem_audit_tools/__main__.py"]
stdout: str
with mock.patch.object(sys, "argv", argv):
stdout = self.__run_main(["--help"])[1]
self.assertIn(
"usage: python -m pop_fem_audit_tools ", stdout)
def test_usage_prog_console_script(self) -> None:
"""Test the usage program name when run as a script."""
argv: list[str] = ["/x/bin/pop-fem-audit-tools"]
stdout: str
with mock.patch.object(sys, "argv", argv):
stdout = self.__run_main(["--help"])[1]
self.assertIn("usage: pop-fem-audit-tools ", stdout)
def test_subcommand_prog_module_run(self) -> None:
"""Test the subcommand program name with python -m."""
argv: list[str] = ["/x/pop_fem_audit_tools/__main__.py"]
self.assertEqual(
self.__dispatched_prog(argv),
"python -m pop_fem_audit_tools run-llm")
def test_subcommand_prog_console_script(self) -> None:
"""Test the subcommand program name as a script."""
argv: list[str] = ["/x/bin/pop-fem-audit-tools"]
self.assertEqual(
self.__dispatched_prog(argv),
"pop-fem-audit-tools run-llm")
def __dispatched_prog(self, argv: list[str]) -> str:
"""Run a stub run-llm tool and return the prog it sees.
:param argv: The value to patch ``sys.argv`` with.
:return: The default argparse program name seen by the
dispatched tool.
"""
seen: list[str] = []
def tool(tool_argv: list[str] | None) -> int:
"""Record the argparse program name and succeed.
:param tool_argv: The tool command-line arguments.
:return: Always 0.
"""
parser: argparse.ArgumentParser = \
argparse.ArgumentParser()
seen.append(parser.prog)
return 0
with mock.patch.object(sys, "argv", argv), \
mock.patch.dict(__main__.SUBCOMMANDS,
{"run-llm": tool}):
self.__run_main(["run-llm"])
return seen[0]
def test_unknown_subcommand_exits_non_zero(self) -> None:
"""Test that an unknown subcommand errors to standard error."""
status: int
stdout: str
stderr: str
status, stdout, stderr = self.__run_main(["nonsense"])
self.assertNotEqual(status, 0)
self.assertIn("nonsense", stderr)
tools/tests/test_run_llm.py
+557
View File
@@ -0,0 +1,557 @@
# Tools for A Feminist Audit of Pop Music.
# Copyright 2026 imacat. All rights reserved.
# Authors:
# imacat@mail.imacat.idv.tw (imacat), 2026/7/30
# AI assistance: Claude Code (Anthropic)
"""Unit tests for the run_llm batch runner module."""
import io
import json
import os
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
from datetime import datetime
from pathlib import Path
from typing import Any
from unittest import mock
from pop_fem_audit_tools import run_llm
class RunLLMTestCase(unittest.TestCase):
"""The common base test case with the shared helpers."""
def _make_temp_dir(self) -> Path:
"""Create a temporary directory removed on test cleanup.
:return: The path of the temporary directory.
"""
tmp: tempfile.TemporaryDirectory[str] \
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
return Path(tmp.name)
def _make_success_entry(self, custom_id: str,
text: str) -> mock.Mock:
"""Create a mock succeeded batch result entry.
:param custom_id: The custom ID of the entry.
:param text: The output text.
:return: The mock result entry.
"""
entry: mock.Mock = mock.Mock()
entry.custom_id = custom_id
entry.result = mock.Mock()
entry.result.type = "succeeded"
entry.result.message = self.__make_message(text)
return entry
@staticmethod
def _make_error_entry(custom_id: str,
error_type: str) -> mock.Mock:
"""Create a mock errored batch result entry.
:param custom_id: The custom ID of the entry.
:param error_type: The error type.
:return: The mock result entry.
"""
entry: mock.Mock = mock.Mock()
entry.custom_id = custom_id
entry.result = mock.Mock()
entry.result.type = "errored"
entry.result.error = mock.Mock()
entry.result.error.type = error_type
return entry
@staticmethod
def __make_message(text: str) -> mock.Mock:
"""Create a mock message with a single text block.
:param text: The text of the text block.
:return: The mock message.
"""
block: mock.Mock = mock.Mock()
block.type = "text"
block.text = text
usage: mock.Mock = mock.Mock()
usage.model_dump.return_value = {
"input_tokens": 10, "output_tokens": 5, "extra": None}
message: mock.Mock = mock.Mock()
message.content = [block]
message.stop_reason = "end_turn"
message.usage = usage
return message
class TestLoadItems(RunLLMTestCase):
"""Test cases for the input JSONL validation."""
def setUp(self) -> None:
"""Create a temporary directory for the input files."""
self.__dir: Path = self._make_temp_dir()
def __write_input(self, content: str) -> Path:
"""Write an input file with the given content.
:param content: The file content.
:return: The path of the input file.
"""
path: Path = self.__dir / "items.jsonl"
path.write_text(content, encoding="utf-8")
return path
def test_valid_items(self) -> None:
"""Test that valid items are loaded in file order."""
path: Path = self.__write_input(
'{"id": "a", "content": "one"}\n'
'{"id": "b", "content": "two"}\n')
items: list[run_llm.Item] = run_llm.load_items(path)
self.assertEqual(items, [{"id": "a", "content": "one"},
{"id": "b", "content": "two"}])
def test_malformed_json_names_line(self) -> None:
"""Test that malformed JSON reports the line number."""
path: Path = self.__write_input(
'{"id": "a", "content": "one"}\n'
'not json\n')
with self.assertRaises(run_llm.InputFormatError) as context:
run_llm.load_items(path)
self.assertIn("line 2", str(context.exception))
def test_missing_key_names_line(self) -> None:
"""Test that a missing key reports the line number."""
path: Path = self.__write_input('{"id": "a"}\n')
with self.assertRaises(run_llm.InputFormatError) as context:
run_llm.load_items(path)
self.assertIn("line 1", str(context.exception))
def test_extra_key_rejected(self) -> None:
"""Test that an extra key is rejected."""
path: Path = self.__write_input(
'{"id": "a", "content": "one", "extra": 1}\n')
with self.assertRaises(run_llm.InputFormatError):
run_llm.load_items(path)
def test_non_string_content_rejected(self) -> None:
"""Test that a non-string content is rejected."""
path: Path = self.__write_input('{"id": "a", "content": 3}\n')
with self.assertRaises(run_llm.InputFormatError):
run_llm.load_items(path)
def test_duplicated_id_names_line(self) -> None:
"""Test that a duplicated ID reports the line number."""
path: Path = self.__write_input(
'{"id": "a", "content": "one"}\n'
'{"id": "a", "content": "two"}\n')
with self.assertRaises(run_llm.InputFormatError) as context:
run_llm.load_items(path)
self.assertIn("line 2", str(context.exception))
self.assertIn("a", str(context.exception))
def test_empty_file_rejected(self) -> None:
"""Test that an empty input file is rejected."""
path: Path = self.__write_input("")
with self.assertRaises(run_llm.InputFormatError):
run_llm.load_items(path)
class TestEnvParsing(RunLLMTestCase):
"""Test cases for the .env parsing and API key resolution."""
def setUp(self) -> None:
"""Create a temporary directory for the .env files."""
self.__dir: Path = self._make_temp_dir()
def test_parse_env_file(self) -> None:
"""Test parsing a .env file with comments and blanks."""
path: Path = self.__dir / ".env"
path.write_text(
"# a comment\n"
"\n"
"ANTHROPIC_API_KEY=sk-test-123\n"
"OTHER = value \n"
"garbage line\n",
encoding="utf-8")
values: dict[str, str] = run_llm.parse_env_file(path)
self.assertEqual(values,
{"ANTHROPIC_API_KEY": "sk-test-123",
"OTHER": "value"})
def test_parse_missing_env_file(self) -> None:
"""Test that a missing .env file yields no values."""
values: dict[str, str] = run_llm.parse_env_file(
self.__dir / ".env")
self.assertEqual(values, {})
def test_resolve_from_environment(self) -> None:
"""Test that the environment variable takes precedence."""
path: Path = self.__dir / ".env"
path.write_text("ANTHROPIC_API_KEY=sk-file\n",
encoding="utf-8")
with mock.patch.dict(os.environ,
{"ANTHROPIC_API_KEY": "sk-env"}):
self.assertEqual(run_llm.resolve_api_key(path), "sk-env")
def test_resolve_from_env_file(self) -> None:
"""Test that the .env file is used as a fallback."""
path: Path = self.__dir / ".env"
path.write_text("ANTHROPIC_API_KEY=sk-file\n",
encoding="utf-8")
with mock.patch.dict(os.environ, {}, clear=True):
self.assertEqual(run_llm.resolve_api_key(path), "sk-file")
def test_resolve_missing_key(self) -> None:
"""Test that a missing API key raises an error."""
with mock.patch.dict(os.environ, {}, clear=True):
with self.assertRaises(RuntimeError):
run_llm.resolve_api_key(self.__dir / ".env")
class TestRequestBuilding(RunLLMTestCase):
"""Test cases for the request construction."""
def test_build_request(self) -> None:
"""Test the shape of a batch request."""
request: dict[str, Any] = run_llm.build_request(
{"id": "song-1", "content": "the lyrics"},
"the system prompt", 2048)
self.assertEqual(request["custom_id"], "song-1")
params: dict[str, Any] = request["params"]
self.assertEqual(params["model"], "claude-sonnet-4-6")
self.assertEqual(params["temperature"], 0.0)
self.assertEqual(params["thinking"], {"type": "disabled"})
self.assertEqual(params["max_tokens"], 2048)
self.assertEqual(params["system"], "the system prompt")
self.assertEqual(params["messages"],
[{"role": "user", "content": "the lyrics"}])
def test_arbitration_content(self) -> None:
"""Test the arbitration user message format."""
content: str = run_llm.build_arbitration_content(
"the item", "output one", "output two")
self.assertEqual(content,
"<item>\nthe item\n</item>\n"
"<run1>\noutput one\n</run1>\n"
"<run2>\noutput two\n</run2>")
class TestAgreement(RunLLMTestCase):
"""Test cases for the agreement computation."""
def test_split_by_agreement(self) -> None:
"""Test splitting items into agreed and disagreeing ones."""
items: list[run_llm.Item] = [
{"id": "a", "content": "one"},
{"id": "b", "content": "two"},
{"id": "c", "content": "three"}]
run1: run_llm.Results = {
"a": {"id": "a", "text": "same\n"},
"b": {"id": "b", "text": "left"},
"c": {"id": "c", "text": " padded "}}
run2: run_llm.Results = {
"a": {"id": "a", "text": "same"},
"b": {"id": "b", "text": "right"},
"c": {"id": "c", "text": "padded"}}
agreed: list[str]
disagreed: list[str]
agreed, disagreed = run_llm.split_by_agreement(
items, run1, run2)
self.assertEqual(agreed, ["a", "c"])
self.assertEqual(disagreed, ["b"])
class TestFinalRecords(RunLLMTestCase):
"""Test cases for the final record assembly."""
def test_build_final_records(self) -> None:
"""Test assembling final records from runs and arbitration."""
items: list[run_llm.Item] = [
{"id": "a", "content": "one"},
{"id": "b", "content": "two"}]
run1: run_llm.Results = {
"a": {"id": "a", "text": "agreed text\n"},
"b": {"id": "b", "text": "left"}}
arbitration: run_llm.Results = {
"b": {"id": "b", "text": "arbitrated text"}}
records: list[dict[str, str]] = run_llm.build_final_records(
items, run1, arbitration)
self.assertEqual(records, [
{"id": "a", "text": "agreed text", "source": "agreed"},
{"id": "b", "text": "arbitrated text",
"source": "arbitration"}])
class TestCollectResults(RunLLMTestCase):
"""Test cases for the batch result collection."""
def test_collect_success_and_error(self) -> None:
"""Test collecting succeeded and errored results."""
client: mock.Mock = mock.Mock()
client.messages.batches.results.return_value = iter([
self._make_success_entry("a", "output a"),
self._make_error_entry("b", "invalid_request")])
results: run_llm.Results = run_llm.collect_results(
client, "batch_x")
self.assertEqual(results["a"]["text"], "output a")
self.assertEqual(results["a"]["stop_reason"], "end_turn")
self.assertEqual(results["a"]["usage"],
{"input_tokens": 10, "output_tokens": 5})
self.assertEqual(results["b"],
{"id": "b", "error": "invalid_request"})
client.messages.batches.results.assert_called_once_with(
"batch_x")
def test_find_failures(self) -> None:
"""Test finding failed and missing items."""
results: run_llm.Results = {
"a": {"id": "a", "text": "fine"},
"b": {"id": "b", "error": "errored"}}
self.assertEqual(
run_llm.find_failures(["a", "b", "c"], results),
["b", "c"])
class TestArchive(RunLLMTestCase):
"""Test cases for the archive directory handling."""
def setUp(self) -> None:
"""Create a temporary directory as the runs root."""
self.__dir: Path = self._make_temp_dir()
def test_create_archive_dir(self) -> None:
"""Test the archive directory naming and creation."""
now: datetime = datetime(2026, 7, 30, 20, 5)
directory: Path = run_llm.create_archive_dir(
self.__dir, "coding", Path("prompts/gender_v3.md"), now)
self.assertTrue(directory.is_dir())
self.assertEqual(
directory,
self.__dir / "coding" / "20260730-2005-gender_v3")
def test_existing_archive_dir_rejected(self) -> None:
"""Test that an existing archive directory is rejected."""
now: datetime = datetime(2026, 7, 30, 20, 5)
run_llm.create_archive_dir(
self.__dir, "coding", Path("gender_v3.md"), now)
with self.assertRaises(FileExistsError):
run_llm.create_archive_dir(
self.__dir, "coding", Path("gender_v3.md"), now)
def test_write_jsonl(self) -> None:
"""Test writing records as JSON Lines."""
path: Path = self.__dir / "out.jsonl"
run_llm.write_jsonl(path, [{"id": "a", "text": "中文"},
{"id": "b", "text": "two"}])
lines: list[str] = path.read_text(
encoding="utf-8").splitlines()
self.assertEqual(len(lines), 2)
self.assertEqual(json.loads(lines[0]),
{"id": "a", "text": "中文"})
self.assertIn("中文", lines[0])
class TestMainFlow(RunLLMTestCase):
"""Test cases for the end-to-end main flow."""
def setUp(self) -> None:
"""Create a temporary working directory with input files."""
directory: Path = self._make_temp_dir()
old_cwd: str = os.getcwd()
self.addCleanup(os.chdir, old_cwd)
os.chdir(directory)
Path("prompts").mkdir()
Path("prompts/task_v1.md").write_text(
"The task prompt.\n", encoding="utf-8")
Path("prompts/task_arbitration_v1.md").write_text(
"The arbitration prompt.\n", encoding="utf-8")
Path("items.jsonl").write_text(
'{"id": "a", "content": "first item"}\n'
'{"id": "b", "content": "second item"}\n',
encoding="utf-8")
self.__argv: list[str] = [
"--prompt", "prompts/task_v1.md",
"--arbitration-prompt", "prompts/task_arbitration_v1.md",
"--input", "items.jsonl",
"--phase", "coding"]
@staticmethod
def __make_client(run1: list[Any], run2: list[Any],
arbitration: list[Any] | None = None) \
-> mock.Mock:
"""Create a mock Anthropic client serving canned batch results.
:param run1: The result entries of the run-1 batch.
:param run2: The result entries of the run-2 batch.
:param arbitration: The result entries of the arbitration
batch.
:return: The mock client.
"""
client: mock.Mock = mock.Mock()
batch_ids: list[str] = ["batch_run1", "batch_run2",
"batch_arb"]
client.messages.batches.create.side_effect = [
mock.Mock(id=x) for x in batch_ids]
ended: mock.Mock = mock.Mock()
ended.processing_status = "ended"
ended.ended_at = "2026-07-30T20:00:00+08:00"
client.messages.batches.retrieve.return_value = ended
results: dict[str, list[Any]] = {
"batch_run1": run1, "batch_run2": run2,
"batch_arb": arbitration if arbitration is not None
else []}
client.messages.batches.results.side_effect = (
lambda batch_id: iter(results[batch_id]))
return client
@staticmethod
def __run_main(argv: list[str],
client: mock.Mock | None = None) -> tuple[int, str]:
"""Run main() with a mocked client and captured output.
:param argv: The command-line arguments.
:param client: The mock client, or None for dry runs.
:return: A tuple of the exit status and the standard output.
"""
stdout: io.StringIO = io.StringIO()
stderr: io.StringIO = io.StringIO()
environ: dict[str, str] = {"ANTHROPIC_API_KEY": "sk-test"}
with mock.patch.dict(os.environ, environ), \
mock.patch.object(run_llm.anthropic, "Anthropic",
return_value=client), \
redirect_stdout(stdout), redirect_stderr(stderr):
status: int = run_llm.main(argv)
return status, stdout.getvalue()
def __archive_dir(self) -> Path:
"""Locate the single archive directory of the run.
:return: The archive directory.
"""
directories: list[Path] = list(Path("runs/coding").iterdir())
self.assertEqual(len(directories), 1)
return directories[0]
def test_dry_run(self) -> None:
"""Test the dry-run behavior."""
status: int
stdout: str
status, stdout = self.__run_main(
self.__argv + ["--dry-run"])
self.assertEqual(status, 0)
run_dir: Path = self.__archive_dir()
self.assertEqual((run_dir / "prompt.md").read_text(
encoding="utf-8"), "The task prompt.\n")
self.assertEqual(
(run_dir / "arbitration_prompt.md").read_text(
encoding="utf-8"), "The arbitration prompt.\n")
meta: dict[str, Any] = json.loads(
(run_dir / "meta.json").read_text(encoding="utf-8"))
self.assertTrue(meta["dry_run"])
self.assertEqual(meta["item_count"], 2)
self.assertEqual(meta["model"], "claude-sonnet-4-6")
self.assertFalse((run_dir / "run1.jsonl").exists())
request: dict[str, Any] = json.loads(stdout)
self.assertEqual(request["custom_id"], "a")
self.assertEqual(request["params"]["system"],
"The task prompt.\n")
def test_all_agreed_skips_arbitration(self) -> None:
"""Test that full agreement skips the arbitration batch."""
client: mock.Mock = self.__make_client(
run1=[self._make_success_entry("a", "answer a"),
self._make_success_entry("b", "answer b")],
run2=[self._make_success_entry("a", "answer a\n"),
self._make_success_entry("b", "answer b")])
status: int = self.__run_main(self.__argv, client)[0]
self.assertEqual(status, 0)
self.assertEqual(
client.messages.batches.create.call_count, 2)
run_dir: Path = self.__archive_dir()
self.assertEqual(
(run_dir / "arbitration.jsonl").read_text(
encoding="utf-8"), "")
final: list[dict[str, Any]] = [
json.loads(x) for x in (run_dir / "final.jsonl")
.read_text(encoding="utf-8").splitlines()]
self.assertEqual(final, [
{"id": "a", "text": "answer a", "source": "agreed"},
{"id": "b", "text": "answer b", "source": "agreed"}])
meta: dict[str, Any] = json.loads(
(run_dir / "meta.json").read_text(encoding="utf-8"))
self.assertEqual(meta["agreed_count"], 2)
self.assertEqual(meta["agreement_rate"], 1.0)
self.assertNotIn("arbitration", meta["batches"])
def test_disagreement_triggers_arbitration(self) -> None:
"""Test that disagreeing items go through arbitration."""
client: mock.Mock = self.__make_client(
run1=[self._make_success_entry("a", "answer a"),
self._make_success_entry("b", "answer b1")],
run2=[self._make_success_entry("a", "answer a"),
self._make_success_entry("b", "answer b2")],
arbitration=[
self._make_success_entry("b", "answer b final")])
status: int = self.__run_main(self.__argv, client)[0]
self.assertEqual(status, 0)
self.assertEqual(
client.messages.batches.create.call_count, 3)
arb_call: mock.call = \
client.messages.batches.create.call_args_list[2]
arb_requests: list[dict[str, Any]] = \
arb_call.kwargs["requests"]
self.assertEqual(len(arb_requests), 1)
self.assertEqual(arb_requests[0]["custom_id"], "b")
self.assertEqual(arb_requests[0]["params"]["system"],
"The arbitration prompt.\n")
self.assertEqual(
arb_requests[0]["params"]["messages"][0]["content"],
"<item>\nsecond item\n</item>\n"
"<run1>\nanswer b1\n</run1>\n"
"<run2>\nanswer b2\n</run2>")
run_dir: Path = self.__archive_dir()
arb_lines: list[str] = (run_dir / "arbitration.jsonl") \
.read_text(encoding="utf-8").splitlines()
self.assertEqual(len(arb_lines), 1)
self.assertEqual(json.loads(arb_lines[0])["text"],
"answer b final")
final: list[dict[str, Any]] = [
json.loads(x) for x in (run_dir / "final.jsonl")
.read_text(encoding="utf-8").splitlines()]
self.assertEqual(final, [
{"id": "a", "text": "answer a", "source": "agreed"},
{"id": "b", "text": "answer b final",
"source": "arbitration"}])
meta: dict[str, Any] = json.loads(
(run_dir / "meta.json").read_text(encoding="utf-8"))
self.assertEqual(meta["agreed_count"], 1)
self.assertEqual(meta["agreement_rate"], 0.5)
self.assertEqual(meta["batches"]["arbitration"]["batch_id"],
"batch_arb")
def test_run_failure_exits_non_zero(self) -> None:
"""Test that a failed item aborts with a non-zero status."""
client: mock.Mock = self.__make_client(
run1=[self._make_success_entry("a", "answer a"),
self._make_error_entry("b", "invalid_request")],
run2=[self._make_success_entry("a", "answer a"),
self._make_success_entry("b", "answer b")])
status: int = self.__run_main(self.__argv, client)[0]
self.assertEqual(status, 1)
run_dir: Path = self.__archive_dir()
self.assertTrue((run_dir / "run1.jsonl").exists())
self.assertTrue((run_dir / "run2.jsonl").exists())
self.assertTrue((run_dir / "meta.json").exists())
self.assertFalse((run_dir / "final.jsonl").exists())
run1_lines: list[str] = (run_dir / "run1.jsonl") \
.read_text(encoding="utf-8").splitlines()
self.assertEqual(json.loads(run1_lines[1]),
{"id": "b", "error": "invalid_request"})
def test_invalid_input_exits_non_zero(self) -> None:
"""Test that an invalid input file aborts before archiving."""
Path("items.jsonl").write_text(
'{"id": "a"}\n', encoding="utf-8")
status: int = self.__run_main(
self.__argv + ["--dry-run"])[0]
self.assertEqual(status, 1)
self.assertFalse(Path("runs").exists())