diff --git a/README.md b/README.md index 01a0248..8f992c1 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ Hot 100（2016–2025）為例的內容分析。 ## 重現方式 -1. 準備 Python 3.12+ 環境，安裝分析管線套件： +1. 準備 Python 3.14+ 環境，安裝分析管線套件： `pip install -e tools/`。 2. 將 Anthropic API 金鑰寫入 `.env`（格式見 `tools/.env.example`）。 diff --git a/docs/decision_log.md b/docs/decision_log.md index 2fbd36b..c0a1428 100644 --- a/docs/decision_log.md +++ b/docs/decision_log.md @@ -73,3 +73,26 @@ 兩層；取得日期不可考者不假造，僅記可證上界。 - **`run_llm` 改走 pydantic-settings 統一設定**（刪手寫 .env parser），與 `config.py` / `database.py` 一致。 +- **`lyrics` 表併入 `songs.lyrics` nullable 欄位**（資料模型由 + 五表改四表）。理由：1—1 關係在此量級下獨立成表只有正規化 + 慣性，nullable 欄位更簡單；「未取得」以 NULL 表達，語意 + 等價。 + +## 2026-08-01 + +- **引擎獨立性以 PostgreSQL/SQLite 為範圍**（MySQL 的 VARCHAR + 長度限制不處理）。 +- **`build-db` 的重置改用逐表 DELETE，不再 drop/create**： + schema 生命週期歸 migration 管，build-db 只管資料。連帶 + 效果：重置成為純 DML,全程單一交易、驗證通過才 commit—— + 建置失敗時前一版資料完好。首次執行仍以 create_all + （checkfirst）補缺表。 +- **song/artist ID 由 build-db 顯式指派**（首次出現順序 1、2、 + 3…），不依賴 autoincrement——PostgreSQL 的 sequence 在 + DELETE 後不重置，顯式指派讓重建決定性跨引擎成立。 +- **禁止文字 SQL statement，一律經 SQLAlchemy ORM/Core API**。 + 唯一記錄在案的例外：SQLite 的 `PRAGMA foreign_keys=ON` + （官方建議作法，無非文字 API 可用；SQLite 的 FK 旗標為 + 逐連線設定）。裁定其歸屬為連線組態，實作於 `database.py` + 的 `__create_engine`——建 engine 時對 SQLite 註冊 connect + listener，所有消費者全程生效，不再由 build-db 各自註冊。 diff --git a/docs/project_structure.md b/docs/project_structure.md index 2849ac1..8be81a4 100644 --- a/docs/project_structure.md +++ b/docs/project_structure.md @@ -15,6 +15,7 @@ pop-fem-audit/ │ ├── artists_wikidata.csv # 捕捉：Wikidata 快照 │ ├── artists_overrides.csv # 人工核定 / 深度背景 │ ├── lyrics_provenance.csv # 歌詞出處（source + method） +│ ├── lyrics_missing.csv # 歌詞缺漏報表（人工補救用） │ └── lyrics/ # 歌詞 .txt 快取 │ # （gitignored，版權） ├── prompts/ # LLM 定義檔（逐字作為 system prompt） @@ -28,8 +29,15 @@ pop-fem-audit/ │ │ # gitignored；含歌詞全文） │ ├── src/pop_fem_audit_tools/ │ │ ├── __main__.py # 套件 CLI 進入點（分派子命令） +│ │ ├── build_db.py # build the SQLite working store +│ │ │ # from the inputs │ │ ├── config.py # pydantic-settings 設定（.env） │ │ ├── database.py # SQLAlchemy engine / session / Base +│ │ ├── fetch_artists.py # fetch artist metadata from +│ │ │ # Wikidata into the snapshot CSV +│ │ ├── fetch_lyrics.py # fetch missing lyrics from the +│ │ │ # public APIs into data/lyrics/ +│ │ ├── models.py # SQLAlchemy ORM 資料模型 │ │ └── run_llm.py # API runner：2+1 協定、Batch API、 │ │ # 自動寫入 runs/；執行方式 │ │ # pop-fem-audit-tools run-llm diff --git a/docs/research_plan.md b/docs/research_plan.md index 0476b4c..18086ce 100644 --- a/docs/research_plan.md +++ b/docs/research_plan.md @@ -45,10 +45,10 @@ ——此為「可再生仍 commit」的唯一例外，理由：引用穩定性 （十年尺度的環境會腐化）、審稿人零門檻、撰稿期數字變動 可 diff。 -- **資料模型**：`songs`（歌曲實體）、`chart_entries` - （1 歌—N 筆榜單紀錄）、`artists`（歌手實體：Wikidata QID、 - 性別、型態、曲風、國籍）、`song_artists`（M—N 關聯：角色、 - 署名順序）、`lyrics`（1—1）。領域不變量（恰 1000 筆榜單、 +- **資料模型**：`songs`（歌曲實體，含 lyrics nullable 欄位）、 + `chart_entries`（1 歌—N 筆榜單紀錄）、`artists`（歌手實體： + Wikidata QID、性別、型態、曲風、國籍）、`song_artists` + （M—N 關聯：角色、署名順序）。領域不變量（恰 1000 筆榜單、 每歌至少一 primary 歌手等）檢查內建於 `build-db`，違規即 建置失敗。 - **歌手背景防火牆**：歌手背景資料只進人工解讀階段（證據表、 diff --git a/tools/README.rst b/tools/README.rst index e71203d..77f9197 100644 --- a/tools/README.rst +++ b/tools/README.rst @@ -31,6 +31,24 @@ Usage Runs a specific tool command, where "command" can be: +build-db +-------- + +Initialize or rebuild the database from the CSV Billboard ranking table, parse songs and its artists, and output song and artist lists for references. Check ``pop-fem-audit-tools build-db -h`` for complete instructions on its usage. + + +fetch-artists +------------- + +Fetch artist data from Wikidata, for ``build-db`` to merge the fetched data into the database. Check ``pop-fem-audit-tools fetch-artists -h`` for complete instructions on its usage. + + +fetch-lyrics +------------ + +Fetch lyrics, for ``build-db`` to merge the fetched lyrics into the database. Lyrics are not committed into the repository due to copyright issue. Check ``pop-fem-audit-tools fetch-lyrics -h`` for complete instructions on its usage. + + run-llm ------- diff --git a/tools/docs/source/pop_fem_audit_tools.rst b/tools/docs/source/pop_fem_audit_tools.rst index bd4f275..074bccd 100644 --- a/tools/docs/source/pop_fem_audit_tools.rst +++ b/tools/docs/source/pop_fem_audit_tools.rst @@ -4,6 +4,14 @@ pop\_fem\_audit\_tools package Submodules ---------- +pop\_fem\_audit\_tools.build\_db module +--------------------------------------- + +.. automodule:: pop_fem_audit_tools.build_db + :members: + :show-inheritance: + :undoc-members: + pop\_fem\_audit\_tools.config module ------------------------------------ @@ -12,6 +20,38 @@ pop\_fem\_audit\_tools.config module :show-inheritance: :undoc-members: +pop\_fem\_audit\_tools.database module +-------------------------------------- + +.. automodule:: pop_fem_audit_tools.database + :members: + :show-inheritance: + :undoc-members: + +pop\_fem\_audit\_tools.fetch\_artists module +-------------------------------------------- + +.. automodule:: pop_fem_audit_tools.fetch_artists + :members: + :show-inheritance: + :undoc-members: + +pop\_fem\_audit\_tools.fetch\_lyrics module +------------------------------------------- + +.. automodule:: pop_fem_audit_tools.fetch_lyrics + :members: + :show-inheritance: + :undoc-members: + +pop\_fem\_audit\_tools.models module +------------------------------------ + +.. automodule:: pop_fem_audit_tools.models + :members: + :show-inheritance: + :undoc-members: + pop\_fem\_audit\_tools.run\_llm module -------------------------------------- diff --git a/tools/pyproject.toml b/tools/pyproject.toml index 8fe5653..3252f0a 100644 --- a/tools/pyproject.toml +++ b/tools/pyproject.toml @@ -8,7 +8,7 @@ name = "pop-fem-audit-tools" dynamic = ["version"] description = "Tools for A Feminist Audit of Pop Music." readme = "README.rst" -requires-python = ">=3.12" +requires-python = ">=3.14" authors = [ {name = "imacat", email = "imacat@mail.imacat.idv.tw"}, ] @@ -31,8 +31,6 @@ classifiers = [ "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", diff --git a/tools/src/pop_fem_audit_tools/__main__.py b/tools/src/pop_fem_audit_tools/__main__.py index 541fff3..61242ee 100644 --- a/tools/src/pop_fem_audit_tools/__main__.py +++ b/tools/src/pop_fem_audit_tools/__main__.py @@ -15,12 +15,20 @@ from collections.abc import Callable from importlib.machinery import ModuleSpec from types import ModuleType -from pop_fem_audit_tools import run_llm +from pop_fem_audit_tools import ( + build_db, + fetch_artists, + fetch_lyrics, + 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]] = { + "build-db": build_db.main, + "fetch-artists": fetch_artists.main, + "fetch-lyrics": fetch_lyrics.main, "run-llm": run_llm.main, } """The dispatch table from the subcommand name to the tool main.""" diff --git a/tools/src/pop_fem_audit_tools/build_db.py b/tools/src/pop_fem_audit_tools/build_db.py new file mode 100644 index 0000000..1f41dd0 --- /dev/null +++ b/tools/src/pop_fem_audit_tools/build_db.py @@ -0,0 +1,385 @@ +# 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 builder of the SQLite working store. + +Rebuilds the working store from scratch out of the committed +inputs: the year-end chart CSV, the lyrics cache, the Wikidata +artist snapshot, and the manual artist overrides. Missing tables +are created on a fresh store; existing tables are never altered, +as the schema lifecycle belongs to the migrations. Every rebuild +deletes all the rows, loads the data, and validates it in one +transaction, committed only after the data passes the validation +invariants; a failed build leaves the previous store contents +intact. + +The rebuild is deterministic: the builder assigns the song and +artist IDs itself, as 1, 2, 3, ... in the first-occurrence file +order, so the IDs are reproducible across rebuilds on every +database engine, given the frozen input file. + +Run from the repository root; the input paths are relative to the +current working directory. +""" +import argparse +import csv +import re +import sys +from collections.abc import Iterable, Sequence +from pathlib import Path +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from .database import Base, ds +from .models import ( + Artist, + ChartEntry, + Song, + SongArtist, +) + +CHART_CSV: Path = Path("data/yearend_hot100_2016_2025.csv") +"""The year-end chart CSV file.""" +LYRICS_DIR: Path = Path("data/lyrics") +"""The lyrics cache directory.""" +WIKIDATA_CSV: Path = Path("data/artists_wikidata.csv") +"""The Wikidata artist snapshot CSV file.""" +OVERRIDES_CSV: Path = Path("data/artists_overrides.csv") +"""The manual artist override CSV file.""" +YEARS: Sequence[int] = range(2016, 2026) +"""The expected chart years.""" +RANKS_PER_YEAR: int = 100 +"""The expected number of ranks on the chart of each year.""" +ARTIST_FIELDS: dict[str, str] = { + "qid": "wikidata_qid", + "gender": "gender", + "artist_type": "artist_type", + "genre": "genre", + "country": "country", +} +"""The artist CSV columns mapped to the Artist attributes.""" +FEATURING_PATTERN: re.Pattern[str] = re.compile( + r" featuring | feat\. ", re.IGNORECASE) +"""The pattern splitting the primary and featured sides.""" +DELIMITER_PATTERN: re.Pattern[str] = re.compile( + r", | & | \+ |(?i: and | x | with )") +"""The pattern splitting the artist names within a side.""" + + +class BuildError(Exception): + """An error that fails the build.""" + + +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="Rebuild the SQLite working store from the" + " committed inputs.") + return parser.parse_args(argv) + + +def parse_artist_credit(credit: str) -> list[tuple[str, str]]: + """Parse a combined artist credit into artists and roles. + + The credit splits into a primary side and a featured side on + the word "featuring" or "feat.", case-insensitively; without + them, every artist is primary. Each side splits into artist + names on the delimiters ", ", " & ", " + " (literally) and + " and ", " x ", " with " (case-insensitively). + + Known limitation: a compound act name that contains one of the + delimiters is over-split; such cases are corrected later via + the human override layer. + + :param credit: The combined artist credit string. + :return: The (name, role) pairs in credit order, primary side + first, with the role "primary" or "featured". + """ + sides: list[str] = FEATURING_PATTERN.split(credit, maxsplit=1) + pairs: list[tuple[str, str]] = [] + role: str + side: str + for side, role in zip(sides, ("primary", "featured")): + token: str + for token in DELIMITER_PATTERN.split(side): + name: str = token.strip() + if name != "": + pairs.append((name, role)) + return pairs + + +def create_song(session: Session, song_id: int, title: str, + credit: str, artists: dict[str, Artist]) -> Song: + """Create a song with its parsed artist credits. + + The song takes the given ID. A newly created artist takes + the ID following the known artists, in credit order. An + artist duplicated within the credit is kept only at its + first occurrence, with a warning to the standard error. + + :param session: The database session. + :param song_id: The song ID to assign. + :param title: The song title. + :param credit: The combined artist credit string. + :param artists: The known artists by name, updated with the + newly created ones as an observable side effect. + :return: The created song, added to the session. + """ + song: Song = Song(id=song_id, title=title, + artist_credit=credit) + session.add(song) + seen: set[str] = set() + position: int = 0 + name: str + role: str + for name, role in parse_artist_credit(credit): + if name in seen: + print(f"warning: {credit}: duplicated artist" + f" \"{name}\"", file=sys.stderr) + continue + seen.add(name) + if name not in artists: + artists[name] = Artist(id=len(artists) + 1, name=name) + session.add(SongArtist(song=song, artist=artists[name], + role=role, position=position)) + position += 1 + return song + + +def load_chart(session: Session, path: Path) -> None: + """Load the chart CSV into songs, chart entries, and credits. + + A song repeated across the rows is stored once, keyed by its + exact title and artist credit; every row yields one chart + entry. The songs and the artists take the IDs 1, 2, 3, ... + in the first-occurrence row order. + + :param session: The database session. + :param path: The chart CSV file with the columns year, rank, + title, and artist. + :return: None. + :raises OSError: When the file cannot be read. + """ + songs: dict[tuple[str, str], Song] = {} + artists: dict[str, Artist] = {} + with open(path, encoding="utf-8", newline="") as file: + row: dict[str, str] + for row in csv.DictReader(file): + key: tuple[str, str] = (row["title"], row["artist"]) + if key not in songs: + songs[key] = create_song( + session, len(songs) + 1, row["title"], + row["artist"], artists) + session.add(ChartEntry(year=int(row["year"]), + rank=int(row["rank"]), + song=songs[key])) + + +def load_lyrics(session: Session, directory: Path) -> None: + """Load the cached lyrics files into the matching songs. + + A missing directory is skipped. A file whose stem is not an + existing song ID is skipped with a warning to the standard + error. + + :param session: The database session, with the songs flushed. + :param directory: The lyrics cache directory with one + ``.txt`` file per song. + :return: None. + :raises OSError: When a lyrics file cannot be read. + """ + if not directory.is_dir(): + return + for path in sorted(directory.glob("*.txt")): + song: Song | None = None + if path.stem.isdigit(): + song = session.get(Song, int(path.stem)) + if song is None: + print(f"warning: {path}: no song with ID" + f" \"{path.stem}\"", file=sys.stderr) + continue + song.lyrics = path.read_text(encoding="utf-8") + + +def apply_artist_csv(session: Session, path: Path) -> None: + """Apply an artist attribute CSV onto the artist rows. + + Artists match by exact name. Only the non-empty cells are + applied, so a later CSV overrides an earlier one field by + field. The note column is ignored. A missing file is + skipped. + + :param session: The database session, with the artists + flushed. + :param path: The CSV file with the columns name, qid, gender, + artist_type, genre, country, and note. + :return: None. + :raises BuildError: When a name matches no artist. + :raises OSError: When the file cannot be read. + """ + if not path.exists(): + return + with open(path, encoding="utf-8", newline="") as file: + row: dict[str, str] + for row in csv.DictReader(file): + artist: Artist | None = session.scalar( + sa.select(Artist) + .where(Artist.name == row["name"])) + if artist is None: + raise BuildError( + f"{path}: no artist named \"{row['name']}\"") + column: str + attribute: str + for column, attribute in ARTIST_FIELDS.items(): + if row.get(column): + setattr(artist, attribute, row[column]) + + +def find_violations(session: Session, years: Iterable[int], + ranks_per_year: int) -> list[str]: + """Find the invariant violations in the loaded data. + + The invariants: the chart entries cover each expected year + and rank exactly once and nothing else, every song has at + least one primary artist, and every artist name is non-empty. + + :param session: The database session with the loaded data + flushed. + :param years: The expected chart years. + :param ranks_per_year: The expected number of ranks per year. + :return: The violation messages, empty when the data is + valid. + """ + violations: list[str] = [] + expected: set[tuple[int, int]] = { + (year, rank) for year in years + for rank in range(1, ranks_per_year + 1)} + actual: set[tuple[int, int]] = { + (x.year, x.rank) + for x in session.scalars(sa.select(ChartEntry))} + year: int + rank: int + for year, rank in sorted(expected - actual): + violations.append( + f"missing chart entry: year {year} rank {rank}") + for year, rank in sorted(actual - expected): + violations.append( + f"unexpected chart entry: year {year} rank {rank}") + primary_ids: set[int] = set(session.scalars( + sa.select(SongArtist.song_id) + .where(SongArtist.role == "primary"))) + song: Song + for song in session.scalars(sa.select(Song).order_by(Song.id)): + if song.id not in primary_ids: + violations.append( + f"song {song.id} \"{song.title}\" has no primary" + " artist") + artist: Artist + for artist in session.scalars(sa.select(Artist)): + if artist.name.strip() == "": + violations.append( + f"artist {artist.id} has an empty name") + return violations + + +def count_rows(session: Session) -> dict[str, int]: + """Count the loaded rows for the build summary. + + :param session: The database session with the loaded data + flushed. + :return: The counts of the songs, chart entries, artists, + credits, and songs with lyrics, under those keys. + """ + return { + "songs": session.scalar( + sa.select(sa.func.count()).select_from(Song)) or 0, + "chart entries": session.scalar( + sa.select(sa.func.count()) + .select_from(ChartEntry)) or 0, + "artists": session.scalar( + sa.select(sa.func.count()).select_from(Artist)) or 0, + "credits": session.scalar( + sa.select(sa.func.count()) + .select_from(SongArtist)) or 0, + "songs with lyrics": session.scalar( + sa.select(sa.func.count()).select_from(Song) + .where(Song.lyrics.is_not(None))) or 0, + } + + +def prepare_engine(engine: sa.Engine) -> None: + """Prepare a SQLite engine for a build. + + For a file-based SQLite engine, the parent directory of the + database file is created when missing. A non-SQLite engine is + left untouched. + + :param engine: The database engine. + :return: None. + """ + if engine.url.get_backend_name() != "sqlite": + return + database: str | None = engine.url.database + if database is not None and database != ":memory:": + Path(database).parent.mkdir(parents=True, exist_ok=True) + + +def reset_store(session: Session) -> None: + """Delete all the rows from every table of the store. + + :param session: The database session. + :return: None. + """ + model: type[Base] + for model in (SongArtist, ChartEntry, Song, Artist): + session.execute(sa.delete(model)) + + +def main(argv: list[str] | None = None) -> int: + """Rebuild the SQLite working store from the inputs. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The exit status: 0 on success, non-zero on failure. + """ + parse_args(argv) + engine: sa.Engine = ds.engine + prepare_engine(engine) + Base.metadata.create_all(engine) + session: Session = ds.get_db() + counts: dict[str, int] + try: + reset_store(session) + load_chart(session, CHART_CSV) + session.flush() + load_lyrics(session, LYRICS_DIR) + apply_artist_csv(session, WIKIDATA_CSV) + apply_artist_csv(session, OVERRIDES_CSV) + session.flush() + violations: list[str] = find_violations( + session, YEARS, RANKS_PER_YEAR) + if len(violations) > 0: + session.rollback() + for violation in violations: + print(f"error: {violation}", file=sys.stderr) + return 1 + counts = count_rows(session) + session.commit() + except (OSError, BuildError) as error: + session.rollback() + print(f"error: {error}", file=sys.stderr) + return 1 + finally: + session.close() + print("done: " + ", ".join(f"{count} {name}" + for name, count in counts.items()), + file=sys.stderr) + return 0 diff --git a/tools/src/pop_fem_audit_tools/database.py b/tools/src/pop_fem_audit_tools/database.py new file mode 100644 index 0000000..df636ca --- /dev/null +++ b/tools/src/pop_fem_audit_tools/database.py @@ -0,0 +1,106 @@ +# 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 database connection. + +""" +from functools import cached_property +from pathlib import Path +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session + +from .config import Settings, get_settings + + +class Base(DeclarativeBase): + """The base data model.""" + + +class DataSource: + """A data source.""" + + @cached_property + def engine(self) -> sa.Engine: + """Returns the database engine. + + :return: The database engine. + """ + settings: Settings = get_settings() + return self.__create_engine(settings.SQLALCHEMY_DATABASE_URL) + + @cached_property + def __session_local(self) -> sessionmaker: + """Returns the callable to connect and return the database session. + + :return: The callable to connect and return the database session. + """ + return sessionmaker( + autocommit=False, autoflush=False, bind=self.engine) + + def get_db(self) -> Session: + """Connects and returns the database session. + + :return: The database session. + """ + return self.__session_local() + + @classmethod + def __create_engine(cls, url: str) -> sa.Engine: + """Constructs and returns the database engine. + + The foreign key enforcement is enabled on every connection + of a SQLite engine. + + :param url: The SQLAlchemy database URL. + :return: The database engine. + """ + url = cls.__resolve_sqlite_relative_url(url) + engine: sa.Engine + if url == "sqlite://": + engine = sa.create_engine( + url, connect_args={"check_same_thread": False}, + poolclass=sa.StaticPool) + else: + engine = sa.create_engine(url) + if engine.url.get_backend_name() == "sqlite": + sa.event.listen(engine, "connect", + cls.__enable_sqlite_foreign_keys) + return engine + + @staticmethod + def __enable_sqlite_foreign_keys(dbapi_connection: Any, + _: Any) -> None: + """Enables the foreign key enforcement on a new connection. + + :param dbapi_connection: The DBAPI connection. + :param _: The connection record (unused). + :return: None. + """ + cursor: Any = dbapi_connection.cursor() + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + + @staticmethod + def __resolve_sqlite_relative_url(url: str) -> str: + """Resolves the SQLite relative URL to the instance folder. + + :param url: The SQLAlchemy database URL. + :return: The resolved SQLAlchemy database URL. + """ + if not url.startswith("sqlite:///"): + return url + path: Path = Path(url[len("sqlite:///"):]) + if path.is_absolute(): + return url + base: Path = Path(__file__).parent.parent + if base.name == "src": + base = base.parent + path = base / "instance" / path + return f"sqlite:///{path}" + + +ds: DataSource = DataSource() +"""The data source.""" diff --git a/tools/src/pop_fem_audit_tools/fetch_artists.py b/tools/src/pop_fem_audit_tools/fetch_artists.py new file mode 100644 index 0000000..61a9e2e --- /dev/null +++ b/tools/src/pop_fem_audit_tools/fetch_artists.py @@ -0,0 +1,360 @@ +# 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 fetcher of the artist metadata. + +Fetches the metadata of the artists without a snapshot row from +Wikidata into the capture layer: the Wikidata artist snapshot +CSV. The working store is only read, never written; the +``build-db`` subcommand assembles the captured files into the +store on the next rebuild. + +Every fetched row is meant for later human verification: the +description of the search hit is recorded in the note column so +that a bad match can be spotted. A search miss or an error on +one artist is noted on its row and does not fail the run. + +Run from the repository root; the data paths are relative to the +current working directory. +""" +import argparse +import csv +import json +import sys +import time +import urllib.parse +import urllib.request +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from .database import ds +from .models import Artist + +WIKIDATA_CSV: Path = Path("data/artists_wikidata.csv") +"""The Wikidata artist snapshot CSV file.""" +SNAPSHOT_FIELDS: Sequence[str] = ( + "name", "qid", "gender", "artist_type", "genre", "country", + "note") +"""The header columns of the Wikidata artist snapshot CSV file.""" +API_URL: str = "https://www.wikidata.org/w/api.php" +"""The URL of the Wikidata API endpoint.""" +USER_AGENT: str = ("pop-fem-audit-tools" + " (https://github.com/imacat/pop-fem-audit)") +"""The User-Agent header sent on every HTTP request.""" +TIMEOUT: float = 30.0 +"""The timeout of an HTTP request, in seconds.""" +SLEEP_SECONDS: float = 1.0 +"""The delay between consecutive HTTP requests, in seconds.""" +HUMAN_QID: str = "Q5" +"""The Wikidata item ID of "human".""" +GROUP_KEYWORDS: Sequence[str] = ("band", "group", "duo", "trio") +"""The label keywords that suggest a musical ensemble, covering +labels like "boy band" and "girl group".""" + + +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="Fetch the artist metadata from Wikidata" + " into the capture layer.") + return parser.parse_args(argv) + + +class ArtistFetcher: + """A fetcher of artist metadata from Wikidata.""" + + def __init__(self) -> None: + """Construct the fetcher.""" + self.__sent: int = 0 + """The number of the HTTP requests already sent.""" + + def fetch(self, name: str) -> dict[str, str]: + """Fetch the metadata of an artist. + + The note column carries the description of the search hit + for human verification. A search miss yields a row with + the note ``not found``. An HTTP, network, or decoding + error yields a row with what was resolved so far and the + note ``error: ``. + + :param name: The artist name to query with. + :return: The snapshot CSV row of the artist. + """ + row: dict[str, str] = {x: "" for x in SNAPSHOT_FIELDS} + row["name"] = name + try: + hit: tuple[str, str] | None = self.__search(name) + if hit is None: + row["note"] = "not found" + return row + row["qid"], row["note"] = hit + self.__resolve(row) + except (OSError, ValueError) as error: + row["note"] = f"error: {error}" + return row + + def __search(self, name: str) -> tuple[str, str] | None: + """Search Wikidata for an artist. + + :param name: The artist name to search for. + :return: The QID and the description of the first hit, or + None when there is no hit. + :raises OSError: On an HTTP or network error. + :raises ValueError: On a JSON decoding error. + """ + data: Any = self.__get_json({ + "action": "wbsearchentities", "search": name, + "language": "en", "type": "item", "format": "json"}) + hits: Any = data.get("search") \ + if isinstance(data, dict) else None + if not isinstance(hits, list) or len(hits) == 0: + return None + hit: Any = hits[0] + if not isinstance(hit, dict) \ + or not isinstance(hit.get("id"), str): + return None + description: Any = hit.get("description") + return hit["id"], \ + description if isinstance(description, str) else "" + + def __resolve(self, row: dict[str, str]) -> None: + """Resolve the claims of an artist into the row fields. + + :param row: The snapshot CSV row, with the QID set. + :return: None. + :raises OSError: On an HTTP or network error. + :raises ValueError: On a JSON decoding error. + """ + claims: dict[str, list[str]] \ + = self.__get_claims(row["qid"]) + gender_ids: list[str] = claims.get("P21", []) + type_ids: list[str] = claims.get("P31", []) + genre_ids: list[str] = claims.get("P136", []) + country_ids: list[str] = claims.get("P27", []) + if len(country_ids) == 0: + country_ids = claims.get("P495", []) + labels: dict[str, str] = self.__get_labels( + gender_ids + type_ids + genre_ids + country_ids) + if len(gender_ids) > 0: + row["gender"] = labels.get(gender_ids[0], "") + row["artist_type"] = self.__artist_type(type_ids, labels) + row["genre"] = "; ".join( + labels[x] for x in genre_ids if x in labels) + if len(country_ids) > 0: + row["country"] = labels.get(country_ids[0], "") + + def __get_claims(self, qid: str) -> dict[str, list[str]]: + """Fetch the item-ID claim targets of a Wikidata item. + + :param qid: The item ID. + :return: The item-ID targets of the gender, instance-of, + genre, and country properties, keyed by the property + ID. + :raises OSError: On an HTTP or network error. + :raises ValueError: On a JSON decoding error. + """ + data: Any = self.__get_json({ + "action": "wbgetentities", "ids": qid, + "props": "claims", "format": "json"}) + claims: Any = None + if isinstance(data, dict) \ + and isinstance(data.get("entities"), dict) \ + and isinstance(data["entities"].get(qid), dict): + claims = data["entities"][qid].get("claims") + if not isinstance(claims, dict): + return {} + return {x: self.__targets(claims.get(x)) + for x in ("P21", "P31", "P136", "P27", "P495")} + + @staticmethod + def __targets(statements: Any) -> list[str]: + """Extract the item-ID targets of the property statements. + + :param statements: The statements of a property, or None. + :return: The item IDs of the statement targets. + """ + if not isinstance(statements, list): + return [] + ids: list[str] = [] + statement: Any + for statement in statements: + if not isinstance(statement, dict): + continue + snak: Any = statement.get("mainsnak") + if not isinstance(snak, dict): + continue + datavalue: Any = snak.get("datavalue") + if not isinstance(datavalue, dict): + continue + value: Any = datavalue.get("value") + if isinstance(value, dict) \ + and isinstance(value.get("id"), str): + ids.append(value["id"]) + return ids + + def __get_labels(self, qids: Sequence[str]) \ + -> dict[str, str]: + """Resolve item IDs to their English labels in one batch. + + :param qids: The item IDs, duplicates allowed. + :return: The English labels, keyed by the item ID; the + items without an English label are left out. + :raises OSError: On an HTTP or network error. + :raises ValueError: On a JSON decoding error. + """ + unique: list[str] = list(dict.fromkeys(qids)) + if len(unique) == 0: + return {} + data: Any = self.__get_json({ + "action": "wbgetentities", "ids": "|".join(unique), + "props": "labels", "languages": "en", + "format": "json"}) + entities: Any = data.get("entities") \ + if isinstance(data, dict) else None + if not isinstance(entities, dict): + return {} + labels: dict[str, str] = {} + qid: str + for qid in unique: + entity: Any = entities.get(qid) + if not isinstance(entity, dict) \ + or not isinstance(entity.get("labels"), dict): + continue + label: Any = entity["labels"].get("en") + if isinstance(label, dict) \ + and isinstance(label.get("value"), str): + labels[qid] = label["value"] + return labels + + @staticmethod + def __artist_type(type_ids: Sequence[str], + labels: dict[str, str]) -> str: + """Derive the artist type from the instance-of targets. + + :param type_ids: The item IDs of the instance-of targets. + :param labels: The English labels, keyed by the item ID. + :return: ``solo`` for a human, ``group`` for a musical + ensemble, or empty for the human to decide. + """ + if HUMAN_QID in type_ids: + return "solo" + qid: str + for qid in type_ids: + label: str = labels.get(qid, "").lower() + if any(x in label for x in GROUP_KEYWORDS): + return "group" + return "" + + def __get_json(self, params: dict[str, str]) -> Any: + """Send a GET request to the API and return the JSON body. + + Consecutive requests are separated by a fixed delay. + + :param params: The query parameters. + :return: The parsed JSON body. + :raises OSError: On an HTTP or network error. + :raises ValueError: On a JSON decoding error. + """ + if self.__sent > 0: + time.sleep(SLEEP_SECONDS) + self.__sent += 1 + url: str = f"{API_URL}?{urllib.parse.urlencode(params)}" + request: urllib.request.Request = urllib.request.Request( + url, headers={"User-Agent": USER_AGENT}) + with urllib.request.urlopen( + request, timeout=TIMEOUT) as response: + return json.load(response) + + +def read_snapshot_names() -> set[str]: + """Read the artist names already in the snapshot CSV file. + + :return: The artist names, or an empty set when the file is + missing. + :raises OSError: When the file cannot be read. + """ + if not WIKIDATA_CSV.exists(): + return set() + with open(WIKIDATA_CSV, encoding="utf-8", + newline="") as file: + reader: csv.DictReader[str] = csv.DictReader(file) + return {x["name"] for x in reader + if x.get("name") is not None} + + +def append_row(row: dict[str, str]) -> None: + """Append a row to the snapshot CSV file. + + The CSV file is created with the header row when missing; the + existing rows are preserved. + + :param row: The snapshot CSV row. + :return: None. + :raises OSError: When the file cannot be written. + """ + is_new: bool = not WIKIDATA_CSV.exists() + WIKIDATA_CSV.parent.mkdir(parents=True, exist_ok=True) + with open(WIKIDATA_CSV, "a", encoding="utf-8", + newline="") as file: + writer: csv.DictWriter[str] = csv.DictWriter( + file, SNAPSHOT_FIELDS) + if is_new: + writer.writeheader() + writer.writerow(row) + + +def main(argv: list[str] | None = None) -> int: + """Fetch the artist metadata from Wikidata. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The exit status: 0 on success, misses and errors + included, non-zero on a setup error. + """ + parse_args(argv) + fetcher: ArtistFetcher = ArtistFetcher() + fetched: int = 0 + not_found: int = 0 + errors: int = 0 + skipped: int = 0 + session: Session = ds.get_db() + try: + done: set[str] = read_snapshot_names() + name: str + for name in session.scalars( + sa.select(Artist.name).order_by(Artist.id)): + if name in done: + skipped += 1 + continue + row: dict[str, str] = fetcher.fetch(name) + append_row(row) + status: str = row["qid"] + if row["note"] == "not found": + not_found += 1 + status = "not found" + elif row["note"].startswith("error: "): + errors += 1 + status = row["note"] + else: + fetched += 1 + print(f"artist \"{name}\": {status}", + file=sys.stderr) + except (OSError, sa.exc.SQLAlchemyError) as error: + print(f"error: {error}", file=sys.stderr) + return 1 + finally: + session.close() + print(f"done: {fetched} fetched, {not_found} not found," + f" {errors} errors, {skipped} skipped", + file=sys.stderr) + return 0 diff --git a/tools/src/pop_fem_audit_tools/fetch_lyrics.py b/tools/src/pop_fem_audit_tools/fetch_lyrics.py new file mode 100644 index 0000000..1614975 --- /dev/null +++ b/tools/src/pop_fem_audit_tools/fetch_lyrics.py @@ -0,0 +1,283 @@ +# 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 fetcher of the missing song lyrics. + +Fetches the lyrics of the songs without a cache file from the +public lyrics APIs, Lyrics.ovh and LRCLIB, into the capture +layer: the lyrics cache directory and the provenance CSV. The +working store is only read, never written; the ``build-db`` +subcommand assembles the captured files into the store on the +next rebuild. + +A song that every API misses is reported in the missing lyrics +CSV, which is rewritten on every run to reflect the current +status. Misses are expected and do not fail the run. + +Run from the repository root; the data paths are relative to the +current working directory. +""" +import argparse +import csv +import datetime +import json +import sys +import time +import urllib.parse +import urllib.request +from collections.abc import Sequence +from pathlib import Path +from typing import Any + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from .database import ds +from .models import ( + Artist, + Song, + SongArtist, +) + +LYRICS_DIR: Path = Path("data/lyrics") +"""The lyrics cache directory.""" +PROVENANCE_CSV: Path = Path("data/lyrics_provenance.csv") +"""The lyrics provenance CSV file.""" +MISSING_CSV: Path = Path("data/lyrics_missing.csv") +"""The missing lyrics report CSV file.""" +PROVENANCE_FIELDS: Sequence[str] = ( + "song_id", "source", "method", "acquired_at", "note") +"""The header columns of the lyrics provenance CSV file.""" +MISSING_FIELDS: Sequence[str] = ( + "song_id", "title", "artist_credit", "reason") +"""The header columns of the missing lyrics report CSV file.""" +USER_AGENT: str = ("pop-fem-audit-tools" + " (https://github.com/imacat/pop-fem-audit)") +"""The User-Agent header sent on every HTTP request.""" +TIMEOUT: float = 30.0 +"""The timeout of an HTTP request, in seconds.""" +SLEEP_SECONDS: float = 1.0 +"""The delay between consecutive HTTP requests, in seconds.""" + + +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="Fetch the missing song lyrics from the" + " public lyrics APIs into the capture layer.") + return parser.parse_args(argv) + + +class LyricsFetcher: + """A fetcher of song lyrics from the public lyrics APIs.""" + + def __init__(self) -> None: + """Construct the fetcher.""" + self.__sent: int = 0 + """The number of the HTTP requests already sent.""" + + def fetch(self, artist: str, title: str) \ + -> tuple[str, str] | None: + """Fetch the lyrics of a song. + + The APIs are tried in order, Lyrics.ovh and then LRCLIB, + stopping at the first hit. An HTTP or network error on + an API counts as a miss on that API. + + :param artist: The artist name to query with. + :param title: The song title to query with. + :return: The lyrics text and the source name, or None + when every API misses. + """ + lyrics: str | None = self.__fetch_ovh(artist, title) + if lyrics is not None: + return lyrics, "lyrics.ovh" + lyrics = self.__fetch_lrclib(artist, title) + if lyrics is not None: + return lyrics, "lrclib" + return None + + def __fetch_ovh(self, artist: str, title: str) -> str | None: + """Fetch the lyrics of a song from Lyrics.ovh. + + :param artist: The artist name to query with. + :param title: The song title to query with. + :return: The lyrics text, or None on a miss. + """ + url: str = ("https://api.lyrics.ovh/v1/" + f"{urllib.parse.quote(artist, safe='')}/" + f"{urllib.parse.quote(title, safe='')}") + data: Any = self.__get_json(url) + if isinstance(data, dict) \ + and isinstance(data.get("lyrics"), str) \ + and data["lyrics"] != "": + return data["lyrics"] + return None + + def __fetch_lrclib(self, artist: str, title: str) \ + -> str | None: + """Fetch the lyrics of a song from LRCLIB. + + :param artist: The artist name to query with. + :param title: The song title to query with. + :return: The plain lyrics text, or None on a miss. + """ + query: str = urllib.parse.urlencode( + {"artist_name": artist, "track_name": title}) + url: str = f"https://lrclib.net/api/get?{query}" + data: Any = self.__get_json(url) + if isinstance(data, dict) \ + and isinstance(data.get("plainLyrics"), str) \ + and data["plainLyrics"] != "": + return data["plainLyrics"] + return None + + def __get_json(self, url: str) -> Any: + """Send a GET request and return the parsed JSON body. + + Consecutive requests are separated by a fixed delay. + + :param url: The URL to request. + :return: The parsed JSON body, or None on any HTTP, + network, or decoding error. + """ + if self.__sent > 0: + time.sleep(SLEEP_SECONDS) + self.__sent += 1 + request: urllib.request.Request = urllib.request.Request( + url, headers={"User-Agent": USER_AGENT}) + try: + with urllib.request.urlopen( + request, timeout=TIMEOUT) as response: + return json.load(response) + except (OSError, ValueError): + return None + + +def query_artist(session: Session, song_id: int) -> str | None: + """Find the artist name to query the APIs with. + + :param session: The database session. + :param song_id: The song ID. + :return: The name of the primary-role artist with the lowest + position, or None when the song has no primary artist. + """ + return session.scalar( + sa.select(Artist.name) + .join(SongArtist, SongArtist.artist_id == Artist.id) + .where(SongArtist.song_id == song_id, + SongArtist.role == "primary") + .order_by(SongArtist.position) + .limit(1)) + + +def save_lyrics(song_id: int, lyrics: str) -> None: + """Write the lyrics of a song into the cache directory. + + The cache directory is created when missing. + + :param song_id: The song ID. + :param lyrics: The lyrics text. + :return: None. + :raises OSError: When the file cannot be written. + """ + LYRICS_DIR.mkdir(parents=True, exist_ok=True) + (LYRICS_DIR / f"{song_id}.txt").write_text( + lyrics, encoding="utf-8") + + +def append_provenance(song_id: int, source: str) -> None: + """Append a provenance row for a fetched lyrics file. + + The CSV file is created with the header row when missing. + + :param song_id: The song ID. + :param source: The source name of the fetched lyrics. + :return: None. + :raises OSError: When the file cannot be written. + """ + is_new: bool = not PROVENANCE_CSV.exists() + PROVENANCE_CSV.parent.mkdir(parents=True, exist_ok=True) + with open(PROVENANCE_CSV, "a", encoding="utf-8", + newline="") as file: + writer: Any = csv.writer(file) + if is_new: + writer.writerow(PROVENANCE_FIELDS) + writer.writerow([song_id, source, "api-fetch", + datetime.date.today().isoformat(), ""]) + + +def write_missing(misses: Sequence[Sequence[Any]]) -> None: + """Rewrite the missing lyrics report CSV file. + + The previous content is replaced, so the file reflects the + current misses only. + + :param misses: The rows of the songs still without lyrics, + each with the song ID, the title, the artist credit, and + the reason. + :return: None. + :raises OSError: When the file cannot be written. + """ + MISSING_CSV.parent.mkdir(parents=True, exist_ok=True) + with open(MISSING_CSV, "w", encoding="utf-8", + newline="") as file: + writer: Any = csv.writer(file) + writer.writerow(MISSING_FIELDS) + writer.writerows(misses) + + +def main(argv: list[str] | None = None) -> int: + """Fetch the missing song lyrics from the public APIs. + + :param argv: The command-line arguments, or None for + ``sys.argv``. + :return: The exit status: 0 on success, misses included, + non-zero on a setup error. + """ + parse_args(argv) + fetcher: LyricsFetcher = LyricsFetcher() + fetched: int = 0 + misses: list[tuple[int, str, str, str]] = [] + session: Session = ds.get_db() + try: + song: Song + for song in session.scalars( + sa.select(Song).order_by(Song.id)): + if (LYRICS_DIR / f"{song.id}.txt").exists(): + continue + artist: str | None = query_artist(session, song.id) + result: tuple[str, str] | None = None + reason: str = "no primary artist" + if artist is not None: + result = fetcher.fetch(artist, song.title) + reason = "not found" + if result is None: + misses.append((song.id, song.title, + song.artist_credit, reason)) + print(f"song {song.id} \"{song.title}\": miss", + file=sys.stderr) + continue + lyrics: str + source: str + lyrics, source = result + save_lyrics(song.id, lyrics) + append_provenance(song.id, source) + fetched += 1 + print(f"song {song.id} \"{song.title}\": {source}", + file=sys.stderr) + write_missing(misses) + except (OSError, sa.exc.SQLAlchemyError) as error: + print(f"error: {error}", file=sys.stderr) + return 1 + finally: + session.close() + print(f"done: {fetched} fetched, {len(misses)} missed", + file=sys.stderr) + return 0 diff --git a/tools/src/pop_fem_audit_tools/models.py b/tools/src/pop_fem_audit_tools/models.py new file mode 100644 index 0000000..66d98e3 --- /dev/null +++ b/tools/src/pop_fem_audit_tools/models.py @@ -0,0 +1,103 @@ +# 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 data models. + +The schema covers the year-end chart data: songs with their +lyrics, their yearly chart entries, the individual artists, and +the song-artist credits with the role and order. + +""" +import sqlalchemy as sa +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from .database import Base + + +class Song(Base): + """A song, identified by its title and combined artist credit.""" + __tablename__ = "songs" + """The table name.""" + + id: Mapped[int] = mapped_column(primary_key=True) + """The song ID.""" + title: Mapped[str] = mapped_column() + """The song title.""" + artist_credit: Mapped[str] = mapped_column() + """The combined artist credit string as printed on the chart.""" + lyrics: Mapped[str | None] + """The lyrics text, when available.""" + chart_entries: Mapped[list[ChartEntry]] \ + = relationship(back_populates="song") + """The chart entries of the song.""" + song_artists: Mapped[list[SongArtist]] \ + = relationship(back_populates="song") + """The song-artist credits of the song.""" + __table_args__ = (sa.UniqueConstraint(title, artist_credit),) + """The table-level constraints.""" + + +class ChartEntry(Base): + """An entry of a song on the year-end chart.""" + __tablename__ = "chart_entries" + """The table name.""" + + year: Mapped[int] = mapped_column(primary_key=True) + """The chart year.""" + rank: Mapped[int] = mapped_column(primary_key=True) + """The rank of the song on the chart of the year.""" + song_id: Mapped[int] = mapped_column(sa.ForeignKey(Song.id)) + """The ID of the charted song.""" + song: Mapped[Song] = relationship(back_populates="chart_entries") + """The charted song.""" + + +class Artist(Base): + """An individual artist.""" + __tablename__ = "artists" + """The table name.""" + + id: Mapped[int] = mapped_column(primary_key=True) + """The artist ID.""" + name: Mapped[str] = mapped_column(unique=True) + """The artist name.""" + wikidata_qid: Mapped[str | None] + """The Wikidata QID of the artist.""" + gender: Mapped[str | None] + """The gender of the artist.""" + artist_type: Mapped[str | None] + """The artist type: solo, group, or mixed.""" + genre: Mapped[str | None] + """The music genre of the artist.""" + country: Mapped[str | None] + """The country of the artist.""" + song_artists: Mapped[list[SongArtist]] \ + = relationship(back_populates="artist") + """The song-artist credits of the artist.""" + + +class SongArtist(Base): + """A credit of an artist on a song, with the role and order.""" + __tablename__ = "song_artists" + """The table name.""" + + song_id: Mapped[int] = mapped_column(sa.ForeignKey(Song.id), + primary_key=True) + """The ID of the credited song.""" + artist_id: Mapped[int] = mapped_column( + sa.ForeignKey(Artist.id), primary_key=True) + """The ID of the credited artist.""" + role: Mapped[str] = mapped_column() + """The role of the artist: primary or featured.""" + position: Mapped[int] + """The 0-based position of the artist in the credit order.""" + song: Mapped[Song] = relationship(back_populates="song_artists") + """The credited song.""" + artist: Mapped[Artist] \ + = relationship(back_populates="song_artists") + """The credited artist.""" + __table_args__ = ( + sa.CheckConstraint(role.in_(["primary", "featured"]), + name="ck_song_artists_role"),) + """The table-level constraints.""" diff --git a/tools/tests/test_build_db.py b/tools/tests/test_build_db.py new file mode 100644 index 0000000..ff47893 --- /dev/null +++ b/tools/tests/test_build_db.py @@ -0,0 +1,294 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/7/31 +"""Unit tests for the working store builder module.""" +import io +import os +import tempfile +import unittest +from contextlib import redirect_stderr +from pathlib import Path +from typing import Any +from unittest import mock + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from pop_fem_audit_tools import build_db, config +from pop_fem_audit_tools.database import DataSource +from pop_fem_audit_tools.models import ( + Artist, + ChartEntry, + Song, +) + + +class TestParseArtistCredit(unittest.TestCase): + """Test cases for the artist credit parser.""" + + def test_plain_solo(self) -> None: + """Test a plain solo artist credit.""" + self.assertEqual(build_db.parse_artist_credit("Adele"), + [("Adele", "primary")]) + + def test_featuring_with_and(self) -> None: + """Test a featuring credit with an "and" delimiter.""" + self.assertEqual( + build_db.parse_artist_credit( + "Drake featuring Wizkid and Kyla"), + [("Drake", "primary"), + ("Wizkid", "featured"), + ("Kyla", "featured")]) + + def test_comma_and_ampersand(self) -> None: + """Test a credit with comma and ampersand delimiters.""" + self.assertEqual( + build_db.parse_artist_credit( + "Lady Gaga, Bradley Cooper & BloodPop"), + [("Lady Gaga", "primary"), + ("Bradley Cooper", "primary"), + ("BloodPop", "primary")]) + + def test_x_delimiter(self) -> None: + """Test the "x" delimiter.""" + self.assertEqual( + build_db.parse_artist_credit("KAROL G x Nicki Minaj"), + [("KAROL G", "primary"), + ("Nicki Minaj", "primary")]) + + def test_plus_delimiter(self) -> None: + """Test the "+" delimiter.""" + self.assertEqual( + build_db.parse_artist_credit("Marshmello + Halsey"), + [("Marshmello", "primary"), + ("Halsey", "primary")]) + + def test_with_delimiter(self) -> None: + """Test the "with" delimiter.""" + self.assertEqual( + build_db.parse_artist_credit( + "Kane Brown with Lauren Alaina"), + [("Kane Brown", "primary"), + ("Lauren Alaina", "primary")]) + + def test_feat_abbreviation(self) -> None: + """Test that "Feat." splits the featured side.""" + self.assertEqual( + build_db.parse_artist_credit( + "Ariana Grande Feat. Doja Cat" + " & Megan Thee Stallion"), + [("Ariana Grande", "primary"), + ("Doja Cat", "featured"), + ("Megan Thee Stallion", "featured")]) + + def test_case_insensitive_featuring(self) -> None: + """Test that "Featuring" splits case-insensitively.""" + self.assertEqual( + build_db.parse_artist_credit( + "24kGoldn Featuring iann dior"), + [("24kGoldn", "primary"), + ("iann dior", "featured")]) + + +class TestBuildDB(unittest.TestCase): + """Test cases for the working store build.""" + + CHART_CSV: str = ( + "year,rank,title,artist\n" + "2016,1,Hello,Adele\n" + "2016,2,One Dance,Drake featuring Wizkid\n" + "2017,1,One Dance,Drake featuring Wizkid\n" + "2017,2,Shape of You,Ed Sheeran\n") + """The default chart CSV fixture: 2 years with 2 ranks each.""" + + def setUp(self) -> None: + """Create a temporary working directory with the fixtures.""" + tmp: tempfile.TemporaryDirectory[str] \ + = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.__dir: Path = Path(tmp.name) + old_cwd: str = os.getcwd() + self.addCleanup(os.chdir, old_cwd) + os.chdir(self.__dir) + Path("data").mkdir() + self.__write_chart(self.CHART_CSV) + url: str = f"sqlite:///{self.__dir}/store.sqlite3" + config.set_settings(config.Settings( + SQLALCHEMY_DATABASE_URL=url, + ANTHROPIC_API_KEY="test-key")) + self.__ds: DataSource = DataSource() + patchers: list[Any] = [ + mock.patch.object(build_db, "ds", self.__ds), + mock.patch.object(build_db, "YEARS", [2016, 2017]), + mock.patch.object(build_db, "RANKS_PER_YEAR", 2)] + for patcher in patchers: + patcher.start() + self.addCleanup(patcher.stop) + + @staticmethod + def __write_chart(content: str) -> None: + """Write the chart CSV fixture. + + :param content: The CSV content. + :return: None. + """ + Path("data/yearend_hot100_2016_2025.csv").write_text( + content, encoding="utf-8") + + @staticmethod + def __run_build() -> tuple[int, str]: + """Run the build 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 = build_db.main([]) + return status, stderr.getvalue() + + def __session(self) -> Session: + """Open a database session closed on test cleanup. + + :return: The database session. + """ + session: Session = self.__ds.get_db() + self.addCleanup(session.close) + return session + + def __song_titles(self) -> dict[int, str]: + """Read the song titles keyed by their IDs. + + :return: The song titles, keyed by the song IDs. + """ + session: Session = self.__session() + return {x.id: x.title + for x in session.scalars(sa.select(Song))} + + def test_dedup_repeated_song(self) -> None: + """Test that a song repeated across years is stored once.""" + status: int + stderr: str + status, stderr = self.__run_build() + self.assertEqual(status, 0) + session: Session = self.__session() + self.assertEqual( + len(list(session.scalars(sa.select(Song)))), 3) + song: Song | None = session.scalar( + sa.select(Song).where(Song.title == "One Dance")) + assert song is not None + self.assertEqual(sorted((x.year, x.rank) + for x in song.chart_entries), + [(2016, 2), (2017, 1)]) + self.assertEqual([(x.artist.name, x.role, x.position) + for x in song.song_artists], + [("Drake", "primary", 0), + ("Wizkid", "featured", 1)]) + self.assertIn("3 songs", stderr) + self.assertIn("4 chart entries", stderr) + self.assertIn("4 artists", stderr) + self.assertIn("4 credits", stderr) + + def test_first_run_on_fresh_store(self) -> None: + """Test that a build on a fresh store creates the tables.""" + self.assertFalse((self.__dir / "store.sqlite3").exists()) + self.assertEqual(self.__run_build()[0], 0) + session: Session = self.__session() + self.assertEqual( + len(list(session.scalars(sa.select(Song)))), 3) + + def test_deterministic_ids(self) -> None: + """Test that two rebuilds assign the same song IDs.""" + self.assertEqual(self.__run_build()[0], 0) + titles: dict[int, str] = self.__song_titles() + self.assertEqual(titles, {1: "Hello", 2: "One Dance", + 3: "Shape of You"}) + self.assertEqual(self.__run_build()[0], 0) + self.assertEqual(self.__song_titles(), titles) + + def test_failed_build_keeps_previous(self) -> None: + """Test that a failed build keeps the previous contents.""" + self.assertEqual(self.__run_build()[0], 0) + titles: dict[int, str] = self.__song_titles() + self.__write_chart( + "year,rank,title,artist\n" + "2016,1,Hello,Adele\n") + status: int + stderr: str + status, stderr = self.__run_build() + self.assertNotEqual(status, 0) + self.assertIn("year 2016 rank 2", stderr) + self.assertEqual(self.__song_titles(), titles) + session: Session = self.__session() + self.assertEqual( + len(list(session.scalars(sa.select(ChartEntry)))), 4) + + def test_missing_rank_fails(self) -> None: + """Test that a missing rank fails without partial data.""" + self.__write_chart( + "year,rank,title,artist\n" + "2016,1,Hello,Adele\n" + "2016,2,One Dance,Drake featuring Wizkid\n" + "2017,1,One Dance,Drake featuring Wizkid\n") + status: int + stderr: str + status, stderr = self.__run_build() + self.assertNotEqual(status, 0) + self.assertIn("year 2017 rank 2", stderr) + session: Session = self.__session() + self.assertEqual( + list(session.scalars(sa.select(ChartEntry))), []) + self.assertEqual(list(session.scalars(sa.select(Song))), []) + + def test_overrides_apply_over_wikidata(self) -> None: + """Test that the overrides win over the Wikidata snapshot.""" + Path("data/artists_wikidata.csv").write_text( + "name,qid,gender,artist_type,genre,country,note\n" + "Adele,Q2831,female,solo,pop,GB,\n", + encoding="utf-8") + Path("data/artists_overrides.csv").write_text( + "name,qid,gender,artist_type,genre,country,note\n" + "Adele,,,,soul,,manually checked\n", + encoding="utf-8") + self.assertEqual(self.__run_build()[0], 0) + session: Session = self.__session() + artist: Artist | None = session.scalar( + sa.select(Artist).where(Artist.name == "Adele")) + assert artist is not None + self.assertEqual(artist.genre, "soul") + self.assertEqual(artist.gender, "female") + self.assertEqual(artist.wikidata_qid, "Q2831") + self.assertEqual(artist.country, "GB") + + def test_unknown_override_name_fails(self) -> None: + """Test that an unknown override name fails the build.""" + Path("data/artists_overrides.csv").write_text( + "name,qid,gender,artist_type,genre,country,note\n" + "Adel,,female,,,,typo\n", encoding="utf-8") + status: int + stderr: str + status, stderr = self.__run_build() + self.assertNotEqual(status, 0) + self.assertIn("Adel", stderr) + session: Session = self.__session() + self.assertEqual(list(session.scalars(sa.select(Song))), []) + + def test_lyrics_loaded(self) -> None: + """Test loading the lyrics cache into the songs.""" + Path("data/lyrics").mkdir() + Path("data/lyrics/1.txt").write_text( + "Hello, it's me\n", encoding="utf-8") + Path("data/lyrics/999.txt").write_text( + "orphan\n", encoding="utf-8") + status: int + stderr: str + status, stderr = self.__run_build() + self.assertEqual(status, 0) + self.assertIn("999", stderr) + self.assertIn("1 songs with lyrics", stderr) + session: Session = self.__session() + song: Song | None = session.get(Song, 1) + assert song is not None + self.assertEqual(song.title, "Hello") + self.assertEqual(song.lyrics, "Hello, it's me\n") diff --git a/tools/tests/test_fetch_artists.py b/tools/tests/test_fetch_artists.py new file mode 100644 index 0000000..9c449cb --- /dev/null +++ b/tools/tests/test_fetch_artists.py @@ -0,0 +1,310 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/7/31 +"""Unit tests for the artist metadata fetcher module.""" +import csv +import io +import json +import os +import tempfile +import unittest +import urllib.error +from contextlib import redirect_stderr +from pathlib import Path +from typing import Any +from unittest import mock + +from sqlalchemy.orm import Session + +from pop_fem_audit_tools import config, fetch_artists +from pop_fem_audit_tools.database import Base, DataSource +from pop_fem_audit_tools.models import Artist + + +class TestFetchArtists(unittest.TestCase): + """Test cases for the artist metadata fetcher.""" + + HEADER: list[str] = [ + "name", "qid", "gender", "artist_type", "genre", + "country", "note"] + """The expected header row of the snapshot CSV file.""" + + def setUp(self) -> None: + """Create a temporary working directory with the store.""" + tmp: tempfile.TemporaryDirectory[str] \ + = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.__dir: Path = Path(tmp.name) + old_cwd: str = os.getcwd() + self.addCleanup(os.chdir, old_cwd) + os.chdir(self.__dir) + Path("data").mkdir() + url: str = f"sqlite:///{self.__dir}/store.sqlite3" + config.set_settings(config.Settings( + SQLALCHEMY_DATABASE_URL=url, + ANTHROPIC_API_KEY="test-key")) + self.__ds: DataSource = DataSource() + patchers: list[Any] = [ + mock.patch.object(fetch_artists, "ds", self.__ds), + mock.patch.object(fetch_artists, "SLEEP_SECONDS", + 0.0)] + for patcher in patchers: + patcher.start() + self.addCleanup(patcher.stop) + + def __seed(self, names: list[str]) -> None: + """Create the schema and the fixture artists. + + The artist IDs are assigned in list order starting from + 1. + + :param names: The artist names. + :return: None. + """ + Base.metadata.create_all(self.__ds.engine) + session: Session = self.__ds.get_db() + try: + name: str + for name in names: + session.add(Artist(name=name)) + session.commit() + finally: + session.close() + + @staticmethod + def __response(payload: dict[str, Any]) -> mock.MagicMock: + """Build a fake HTTP response with a JSON body. + + :param payload: The JSON payload of the response body. + :return: The fake response, usable as a context manager. + """ + response: mock.MagicMock = mock.MagicMock() + response.__enter__.return_value = response + response.read.return_value \ + = json.dumps(payload).encode("utf-8") + return response + + @staticmethod + def __server_error() -> urllib.error.HTTPError: + """Build an HTTP 500 error. + + :return: The HTTP 500 error. + """ + return urllib.error.HTTPError( + "https://example.com/", 500, + "Internal Server Error", None, None) + + @staticmethod + def __claim(qid: str) -> dict[str, Any]: + """Build a claim statement with an item-ID target. + + :param qid: The item ID of the statement target. + :return: The claim statement. + """ + return {"mainsnak": {"snaktype": "value", + "datavalue": {"value": {"id": qid}}}} + + @staticmethod + def __labels(labels: dict[str, str]) -> dict[str, Any]: + """Build a label query response payload. + + :param labels: The English labels, keyed by the item ID. + :return: The response payload. + """ + return {"entities": { + x: {"labels": {"en": {"value": y}}} + for x, y in labels.items()}} + + @staticmethod + def __run_fetch() -> tuple[int, str]: + """Run the fetcher 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 = fetch_artists.main([]) + return status, stderr.getvalue() + + @staticmethod + def __read_rows(path: Path) -> list[list[str]]: + """Read the rows of a CSV file. + + :param path: The CSV file. + :return: The rows, the header included. + """ + with open(path, encoding="utf-8", newline="") as file: + return list(csv.reader(file)) + + def test_human_artist(self) -> None: + """Test a human artist resolving the full metadata.""" + self.__seed(["Adele"]) + search: dict[str, Any] = {"search": [ + {"id": "Q1", "description": "English singer"}]} + claims: dict[str, Any] = {"entities": {"Q1": {"claims": { + "P21": [self.__claim("Q2")], + "P31": [self.__claim("Q5")], + "P136": [self.__claim("Q3"), self.__claim("Q4")], + "P27": [self.__claim("Q6")]}}}} + labels: dict[str, Any] = self.__labels({ + "Q2": "female", "Q5": "human", "Q3": "pop", + "Q4": "soul music", "Q6": "United Kingdom"}) + urlopen: mock.Mock + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__response(search), + self.__response(claims), + self.__response(labels)]) as urlopen: + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertEqual(status, 0) + self.assertEqual(urlopen.call_count, 3) + request: Any = urlopen.call_args_list[0][0][0] + self.assertEqual(request.get_header("User-agent"), + fetch_artists.USER_AGENT) + urls: list[str] = [x[0][0].full_url + for x in urlopen.call_args_list] + self.assertEqual( + urls[0], + "https://www.wikidata.org/w/api.php" + "?action=wbsearchentities&search=Adele&language=en" + "&type=item&format=json") + self.assertEqual( + urls[1], + "https://www.wikidata.org/w/api.php" + "?action=wbgetentities&ids=Q1&props=claims" + "&format=json") + self.assertEqual( + urls[2], + "https://www.wikidata.org/w/api.php" + "?action=wbgetentities&ids=Q2%7CQ5%7CQ3%7CQ4%7CQ6" + "&props=labels&languages=en&format=json") + rows: list[list[str]] = self.__read_rows( + Path("data/artists_wikidata.csv")) + self.assertEqual(len(rows), 2) + self.assertEqual(rows[0], self.HEADER) + self.assertEqual(rows[1], [ + "Adele", "Q1", "female", "solo", "pop; soul music", + "United Kingdom", "English singer"]) + self.assertIn( + "1 fetched, 0 not found, 0 errors, 0 skipped", + stderr) + + def test_band(self) -> None: + """Test a band resolving the group type and the origin.""" + self.__seed(["BTS"]) + search: dict[str, Any] = {"search": [ + {"id": "Q10", + "description": "South Korean boy band"}]} + claims: dict[str, Any] = {"entities": {"Q10": {"claims": { + "P31": [self.__claim("Q11")], + "P136": [self.__claim("Q12")], + "P495": [self.__claim("Q13")]}}}} + labels: dict[str, Any] = self.__labels({ + "Q11": "boy band", "Q12": "K-pop", + "Q13": "South Korea"}) + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__response(search), + self.__response(claims), + self.__response(labels)]): + status: int = self.__run_fetch()[0] + self.assertEqual(status, 0) + rows: list[list[str]] = self.__read_rows( + Path("data/artists_wikidata.csv")) + self.assertEqual(len(rows), 2) + self.assertEqual(rows[1], [ + "BTS", "Q10", "", "group", "K-pop", "South Korea", + "South Korean boy band"]) + + def test_not_found(self) -> None: + """Test that a search miss writes a not-found row.""" + self.__seed(["Nobody"]) + urlopen: mock.Mock + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__response({"search": []})] + ) as urlopen: + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertEqual(status, 0) + self.assertEqual(urlopen.call_count, 1) + rows: list[list[str]] = self.__read_rows( + Path("data/artists_wikidata.csv")) + self.assertEqual(len(rows), 2) + self.assertEqual(rows[0], self.HEADER) + self.assertEqual(rows[1], [ + "Nobody", "", "", "", "", "", "not found"]) + self.assertIn( + "0 fetched, 1 not found, 0 errors, 0 skipped", + stderr) + + def test_http_error_continues(self) -> None: + """Test that an HTTP error is noted and the run goes on.""" + self.__seed(["Broken", "Nobody"]) + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__server_error(), + self.__response({"search": []})]): + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertEqual(status, 0) + rows: list[list[str]] = self.__read_rows( + Path("data/artists_wikidata.csv")) + self.assertEqual(len(rows), 3) + self.assertEqual(rows[1][:2], ["Broken", ""]) + self.assertTrue(rows[1][6].startswith("error: ")) + self.assertEqual(rows[2], [ + "Nobody", "", "", "", "", "", "not found"]) + self.assertIn( + "0 fetched, 1 not found, 1 errors, 0 skipped", + stderr) + + def test_rerun_skips_existing(self) -> None: + """Test that the snapshot rows are skipped and preserved.""" + self.__seed(["Adele", "Nobody"]) + snapshot: Path = Path("data/artists_wikidata.csv") + old_row: list[str] = [ + "Adele", "Q1", "female", "solo", "pop", + "United Kingdom", "English singer"] + with open(snapshot, "w", encoding="utf-8", + newline="") as file: + writer: Any = csv.writer(file) + writer.writerow(self.HEADER) + writer.writerow(old_row) + urlopen: mock.Mock + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__response({"search": []})] + ) as urlopen: + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertEqual(status, 0) + self.assertEqual(urlopen.call_count, 1) + rows: list[list[str]] = self.__read_rows(snapshot) + self.assertEqual(len(rows), 3) + self.assertEqual(rows[0], self.HEADER) + self.assertNotIn(self.HEADER, rows[1:]) + self.assertEqual(rows[1], old_row) + self.assertEqual(rows[2], [ + "Nobody", "", "", "", "", "", "not found"]) + self.assertIn( + "0 fetched, 1 not found, 0 errors, 1 skipped", + stderr) + + def test_no_store_fails(self) -> None: + """Test that a missing working store fails the run.""" + urlopen: mock.Mock + with mock.patch("urllib.request.urlopen") as urlopen: + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertNotEqual(status, 0) + urlopen.assert_not_called() + self.assertIn("error:", stderr) diff --git a/tools/tests/test_fetch_lyrics.py b/tools/tests/test_fetch_lyrics.py new file mode 100644 index 0000000..7aa9cea --- /dev/null +++ b/tools/tests/test_fetch_lyrics.py @@ -0,0 +1,280 @@ +# Tools for A Feminist Audit of Pop Music. +# Copyright 2026 imacat. All rights reserved. +# Authors: +# imacat@mail.imacat.idv.tw (imacat), 2026/7/31 +"""Unit tests for the lyrics fetcher module.""" +import csv +import io +import json +import os +import tempfile +import unittest +import urllib.error +from contextlib import redirect_stderr +from pathlib import Path +from typing import Any +from unittest import mock + +from sqlalchemy.orm import Session + +from pop_fem_audit_tools import config, fetch_lyrics +from pop_fem_audit_tools.database import Base, DataSource +from pop_fem_audit_tools.models import ( + Artist, + Song, + SongArtist, +) + + +class TestFetchLyrics(unittest.TestCase): + """Test cases for the lyrics fetcher.""" + + PROVENANCE_HEADER: list[str] = [ + "song_id", "source", "method", "acquired_at", "note"] + """The expected header row of the provenance CSV file.""" + MISSING_HEADER: list[str] = [ + "song_id", "title", "artist_credit", "reason"] + """The expected header row of the missing report CSV file.""" + + def setUp(self) -> None: + """Create a temporary working directory with the store.""" + tmp: tempfile.TemporaryDirectory[str] \ + = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.__dir: Path = Path(tmp.name) + old_cwd: str = os.getcwd() + self.addCleanup(os.chdir, old_cwd) + os.chdir(self.__dir) + Path("data").mkdir() + url: str = f"sqlite:///{self.__dir}/store.sqlite3" + config.set_settings(config.Settings( + SQLALCHEMY_DATABASE_URL=url, + ANTHROPIC_API_KEY="test-key")) + self.__ds: DataSource = DataSource() + patchers: list[Any] = [ + mock.patch.object(fetch_lyrics, "ds", self.__ds), + mock.patch.object(fetch_lyrics, "SLEEP_SECONDS", 0.0)] + for patcher in patchers: + patcher.start() + self.addCleanup(patcher.stop) + + def __seed(self, songs: list[tuple[str, str]]) -> None: + """Create the schema and the fixture songs. + + Each song gets a single primary artist at position 0; + the song IDs are assigned in list order starting from 1. + + :param songs: The (title, artist) pairs. + :return: None. + """ + Base.metadata.create_all(self.__ds.engine) + session: Session = self.__ds.get_db() + try: + artists: dict[str, Artist] = {} + title: str + artist: str + for title, artist in songs: + if artist not in artists: + artists[artist] = Artist(name=artist) + song: Song = Song(title=title, + artist_credit=artist) + session.add(song) + session.add(SongArtist(song=song, + artist=artists[artist], + role="primary", + position=0)) + session.commit() + finally: + session.close() + + @staticmethod + def __response(payload: dict[str, Any]) -> mock.MagicMock: + """Build a fake HTTP response with a JSON body. + + :param payload: The JSON payload of the response body. + :return: The fake response, usable as a context manager. + """ + response: mock.MagicMock = mock.MagicMock() + response.__enter__.return_value = response + response.read.return_value \ + = json.dumps(payload).encode("utf-8") + return response + + @staticmethod + def __not_found() -> urllib.error.HTTPError: + """Build an HTTP 404 error. + + :return: The HTTP 404 error. + """ + return urllib.error.HTTPError( + "https://example.com/", 404, "Not Found", None, None) + + @staticmethod + def __run_fetch() -> tuple[int, str]: + """Run the fetcher 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 = fetch_lyrics.main([]) + return status, stderr.getvalue() + + @staticmethod + def __read_rows(path: Path) -> list[list[str]]: + """Read the rows of a CSV file. + + :param path: The CSV file. + :return: The rows, the header included. + """ + with open(path, encoding="utf-8", newline="") as file: + return list(csv.reader(file)) + + def test_ovh_hit(self) -> None: + """Test that a Lyrics.ovh hit writes the cache files.""" + self.__seed([("Hello", "Adele")]) + urlopen: mock.Mock + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__response( + {"lyrics": "Hello, it's me\n"})]) as urlopen: + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertEqual(status, 0) + self.assertEqual(urlopen.call_count, 1) + request: Any = urlopen.call_args[0][0] + self.assertEqual(request.get_header("User-agent"), + fetch_lyrics.USER_AGENT) + self.assertEqual( + Path("data/lyrics/1.txt").read_text(encoding="utf-8"), + "Hello, it's me\n") + rows: list[list[str]] = self.__read_rows( + Path("data/lyrics_provenance.csv")) + self.assertEqual(len(rows), 2) + self.assertEqual(rows[0], self.PROVENANCE_HEADER) + self.assertEqual(rows[1][:3], + ["1", "lyrics.ovh", "api-fetch"]) + self.assertNotEqual(rows[1][3], "") + self.assertEqual(rows[1][4], "") + self.assertIn("1 fetched, 0 missed", stderr) + + def test_lrclib_fallback(self) -> None: + """Test that an ovh miss falls back to an LRCLIB hit.""" + self.__seed([("Hello", "Adele")]) + urlopen: mock.Mock + with mock.patch( + "urllib.request.urlopen", + side_effect=[ + self.__not_found(), + self.__response({"plainLyrics": "Hello\n"})] + ) as urlopen: + status: int = self.__run_fetch()[0] + self.assertEqual(status, 0) + self.assertEqual(urlopen.call_count, 2) + self.assertEqual( + Path("data/lyrics/1.txt").read_text(encoding="utf-8"), + "Hello\n") + rows: list[list[str]] = self.__read_rows( + Path("data/lyrics_provenance.csv")) + self.assertEqual(rows[1][:2], ["1", "lrclib"]) + + def test_both_miss(self) -> None: + """Test that a double miss reports the song as missing.""" + self.__seed([("Hello", "Adele")]) + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__not_found(), + self.__not_found()]): + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertEqual(status, 0) + self.assertFalse(Path("data/lyrics/1.txt").exists()) + self.assertFalse( + Path("data/lyrics_provenance.csv").exists()) + rows: list[list[str]] = self.__read_rows( + Path("data/lyrics_missing.csv")) + self.assertEqual(len(rows), 2) + self.assertEqual(rows[0], self.MISSING_HEADER) + self.assertEqual(rows[1][:3], ["1", "Hello", "Adele"]) + self.assertIn("0 fetched, 1 missed", stderr) + + def test_cached_song_skipped(self) -> None: + """Test that a cached song triggers no HTTP request.""" + self.__seed([("Hello", "Adele")]) + Path("data/lyrics").mkdir() + Path("data/lyrics/1.txt").write_text( + "cached\n", encoding="utf-8") + urlopen: mock.Mock + with mock.patch("urllib.request.urlopen") as urlopen: + status: int = self.__run_fetch()[0] + self.assertEqual(status, 0) + urlopen.assert_not_called() + self.assertEqual( + Path("data/lyrics/1.txt").read_text(encoding="utf-8"), + "cached\n") + rows: list[list[str]] = self.__read_rows( + Path("data/lyrics_missing.csv")) + self.assertEqual(rows, [self.MISSING_HEADER]) + + def test_url_encoding(self) -> None: + """Test the percent-encoding of the artist and title.""" + self.__seed([("What's Up? / Down", "AC/DC & Friends")]) + urlopen: mock.Mock + with mock.patch( + "urllib.request.urlopen", + side_effect=[self.__not_found(), + self.__not_found()]) as urlopen: + self.__run_fetch() + self.assertEqual(urlopen.call_count, 2) + urls: list[str] = [x[0][0].full_url + for x in urlopen.call_args_list] + self.assertEqual( + urls[0], + "https://api.lyrics.ovh/v1/AC%2FDC%20%26%20Friends/" + "What%27s%20Up%3F%20%2F%20Down") + self.assertEqual( + urls[1], + "https://lrclib.net/api/get" + "?artist_name=AC%2FDC+%26+Friends" + "&track_name=What%27s+Up%3F+%2F+Down") + + def test_provenance_single_header(self) -> None: + """Test that the provenance keeps one header across runs.""" + self.__seed([("Hello", "Adele"), + ("Umbrella", "Rihanna")]) + with mock.patch( + "urllib.request.urlopen", + side_effect=[ + self.__response({"lyrics": "one\n"}), + self.__response({"lyrics": "two\n"})]): + self.assertEqual(self.__run_fetch()[0], 0) + provenance: Path = Path("data/lyrics_provenance.csv") + rows: list[list[str]] = self.__read_rows(provenance) + self.assertEqual(len(rows), 3) + self.assertEqual(rows[0], self.PROVENANCE_HEADER) + Path("data/lyrics/2.txt").unlink() + with mock.patch( + "urllib.request.urlopen", + side_effect=[ + self.__response({"lyrics": "two again\n"})]): + self.assertEqual(self.__run_fetch()[0], 0) + rows = self.__read_rows(provenance) + self.assertEqual(len(rows), 4) + self.assertEqual(rows[0], self.PROVENANCE_HEADER) + self.assertNotIn(self.PROVENANCE_HEADER, rows[1:]) + self.assertEqual([x[0] for x in rows[1:]], + ["1", "2", "2"]) + + def test_no_store_fails(self) -> None: + """Test that a missing working store fails the run.""" + urlopen: mock.Mock + with mock.patch("urllib.request.urlopen") as urlopen: + status: int + stderr: str + status, stderr = self.__run_fetch() + self.assertNotEqual(status, 0) + urlopen.assert_not_called() + self.assertIn("error:", stderr) diff --git a/tools/tests/test_models.py b/tools/tests/test_models.py new file mode 100644 index 0000000..5d7692c --- /dev/null +++ b/tools/tests/test_models.py @@ -0,0 +1,95 @@ +# 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 data models.""" +import unittest + +import sqlalchemy as sa +from sqlalchemy.orm import Session + +from pop_fem_audit_tools import config +from pop_fem_audit_tools.database import Base, DataSource +from pop_fem_audit_tools.models import ( + Artist, + ChartEntry, + Song, + SongArtist, +) + + +class TestModels(unittest.TestCase): + """Test cases for the data models.""" + + def setUp(self) -> None: + """Create the schema on an in-memory SQLite database.""" + config.set_settings(config.Settings( + SQLALCHEMY_DATABASE_URL="sqlite://", + ANTHROPIC_API_KEY="test-key")) + self.__ds: DataSource = DataSource() + Base.metadata.create_all(self.__ds.engine) + self.__session: Session = self.__ds.get_db() + self.addCleanup(self.__session.close) + + def __add_song(self) -> None: + """Add a song with chart entries, artists, and lyrics. + + :return: None. + """ + song: Song = Song(title="One Dance", + artist_credit="Drake featuring Wizkid") + song.chart_entries = [ChartEntry(year=2016, rank=4)] + song.song_artists = [ + SongArtist(artist=Artist(name="Drake"), + role="primary", position=0), + SongArtist(artist=Artist(name="Wizkid"), + role="featured", position=1)] + song.lyrics = "Baby, I like your style" + self.__session.add(song) + self.__session.commit() + + def test_song_graph(self) -> None: + """Test reading a song graph back through relationships.""" + self.__add_song() + self.__session.expunge_all() + song: Song | None = self.__session.scalar( + sa.select(Song).where(Song.title == "One Dance")) + assert song is not None + self.assertEqual(song.artist_credit, + "Drake featuring Wizkid") + self.assertEqual([(x.year, x.rank) + for x in song.chart_entries], + [(2016, 4)]) + self.assertEqual([(x.artist.name, x.role, x.position) + for x in song.song_artists], + [("Drake", "primary", 0), + ("Wizkid", "featured", 1)]) + self.assertEqual(song.lyrics, + "Baby, I like your style") + artist: Artist | None = self.__session.scalar( + sa.select(Artist).where(Artist.name == "Wizkid")) + assert artist is not None + self.assertEqual([x.song.title for x in artist.song_artists], + ["One Dance"]) + + def test_duplicated_song_rejected(self) -> None: + """Test that a duplicated title and artist credit fails.""" + self.__add_song() + self.__session.add( + Song(title="One Dance", + artist_credit="Drake featuring Wizkid")) + with self.assertRaises(sa.exc.IntegrityError): + self.__session.commit() + + def test_invalid_role_rejected(self) -> None: + """Test that an invalid song-artist role fails.""" + self.__add_song() + song: Song | None = self.__session.scalar( + sa.select(Song).where(Song.title == "One Dance")) + assert song is not None + self.__session.add( + SongArtist(song=song, artist=Artist(name="Kyla"), + role="cover", position=2)) + with self.assertRaises(sa.exc.IntegrityError): + self.__session.commit()