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

Add the data layer with the build-db, fetch-lyrics, and fetch-artists subcommands

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
imacat
2026-08-04 15:12:22 +08:00
co-authored by Claude Fable 5
parent f9ab79f0c2
commit 8d223177cd
17 changed files with 2320 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/README.rst
+18
View File
@@ -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
-------
tools/docs/source/pop_fem_audit_tools.rst
+40
View File
@@ -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
--------------------------------------
tools/pyproject.toml
+1 -3
View File
@@ -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",
tools/src/pop_fem_audit_tools/__main__.py
+9 -1
View File
@@ -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."""
tools/src/pop_fem_audit_tools/build_db.py
+385
View File
@@ -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
``<song_id>.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
tools/src/pop_fem_audit_tools/database.py
+106
View File
@@ -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."""
tools/src/pop_fem_audit_tools/fetch_artists.py
+360
View File
@@ -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: <reason>``.
: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
tools/src/pop_fem_audit_tools/fetch_lyrics.py
+283
View File
@@ -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
tools/src/pop_fem_audit_tools/models.py
+103
View File
@@ -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."""
tools/tests/test_build_db.py
+294
View File
@@ -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")
tools/tests/test_fetch_artists.py
+310
View File
@@ -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)
tools/tests/test_fetch_lyrics.py
+280
View File
@@ -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)
tools/tests/test_models.py
+95
View File
@@ -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()