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

Derive the performing group genders from their Wikidata members

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
imacat
2026-08-17 22:38:39 +08:00
co-authored by Claude Opus 5
parent bebf85814e
commit f87e3642f6
8 changed files with 1733 additions and 1051 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/pop_fem_audit_tools/commands/build_db.py
+24 -16
View File
@@ -41,9 +41,12 @@ spelling, except for the names listed in
canonical spelling regardless of which variant is seen first.
Once the artists carry their captured attributes, every song
takes a performer gender derived from the genders of the artists
credited on it, primary and featured alike; see
`PerformerGenderDeriver.performer_gender`.
takes a performer gender derived from the genders of the
performing artists credited on it, primary and featured alike. A
credited artist without an artist type -- a label, a brand, or a
producer collective -- is not a performing act: it has no voice,
so its gender is inapplicable, and it takes no part in the
derivation. See `PerformerGenderDeriver.performer_gender`.
On a successful build, two review CSV files, ``songs.csv`` and
``artists.csv``, are (re)written under the given output directory,
@@ -617,11 +620,13 @@ class CaptureImporter:
class PerformerGenderDeriver:
"""The performer-gender job: derives the song-level performer
gender from the genders of the credited artists."""
gender from the genders of the credited artists that are
performing acts, a credited artist without an artist type
taking no part."""
MIXED: str = "mixed"
"""The performer gender of a song whose credited artists do not
all share one gender."""
"""The performer gender of a song whose performing credited
artists do not all share one gender."""
def __init__(self, session: Session) -> None:
"""Initialize the deriver.
@@ -636,33 +641,36 @@ class PerformerGenderDeriver:
Reads the songs back from the database, including any songs
pending in the same session, and sets
``Song.performer_gender`` from the genders of the artists
credited on the song, primary and featured alike (see
`performer_gender`). When the method returns, the derived
performer genders are queryable in the session.
credited on the song, primary and featured alike, that have
an artist type (see `performer_gender`). A credited artist
without an artist type is not a performing act and takes no
part. When the method returns, the derived performer
genders are queryable in the session.
:return: None.
"""
song: Song
for song in self.__session.scalars(sa.select(Song)):
song.performer_gender = self.performer_gender(
[x.artist.gender for x in song.song_artists])
[x.artist.gender for x in song.song_artists
if x.artist.type])
self.__session.flush()
@classmethod
def performer_gender(
cls, genders: Iterable[str | None]) -> str | None:
"""Combine the credited artists' genders into one value.
"""Combine the performing artists' genders into one value.
A gender that is None or empty counts as unknown. Two or
more distinct known genders give ``MIXED``, an unknown one
notwithstanding, as an unknown cannot undo a disagreement.
A single known gender shared by every credited artist gives
A single known gender shared by every given artist gives
that gender. Anything else -- a single known gender
alongside an unknown one, or no known gender at all -- gives
None.
alongside an unknown one, no known gender at all, or no
gender given at all -- gives None.
:param genders: The genders of the artists credited on one
song, in any order.
:param genders: The genders of the performing artists
credited on one song, in any order.
:return: The performer gender of the song, or None when it
is undetermined.
"""
tools/src/pop_fem_audit_tools/commands/fetch_artists.py
+310 -42
View File
@@ -4,29 +4,33 @@
# 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, given as the positional command-line argument. The working
store is only read, never written; the ``build-db`` subcommand
assembles the captured files into the store on the next rebuild.
Fetches the metadata of the artists that the Wikidata artist
snapshot CSV, given as the positional command-line argument, does
not resolve yet -- the artists without a row, and the artists
whose row has a blank gender -- from Wikidata into the capture
layer. 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 resolved item is recorded in the note column
so that a bad match can be spotted. An unresolved artist or an
error on one artist is noted on its row and does not fail the
run.
run. A row whose name is no longer an artist of the store is
dropped from the snapshot and reported on the standard error.
"""
import argparse
import csv
import enum
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Sequence
from collections.abc import Container, Sequence
from dataclasses import asdict, dataclass, field, fields
from pathlib import Path
from typing import Any, Literal, TextIO
@@ -83,18 +87,23 @@ NOTE_NOT_FOUND: str = "not found"
"""The note sentinel of an artist without a resolved Wikidata
item, written to the snapshot and read back for the
classification."""
PINNED_QIDS: dict[str, str] = {
"Pinkfong": "Q55735607",
}
CORPUS_START_YEAR: int = 2016
"""The first year of the corpus window: a member who left a group
before it never performed a corpus song."""
MIXED_GENDER: str = "mixed"
"""The gender recorded for a group whose members do not share one
gender."""
TIME_YEAR_PATTERN: re.Pattern[str] = re.compile(r"^[+-]?\d+")
"""The leading year of a Wikidata time value."""
PINNED_QIDS: dict[str, str] = {}
"""The last-resort pinned item IDs, keyed by the artist name.
Each entry is for an artist the algorithm documented on
An entry is for an artist the algorithm documented on
``ArtistFetcher`` is structurally unable to resolve, with its
justification recorded here:
- "Pinkfong": the only charting act whose item is typed as a
brand (P31 = Q431289), which the type gate (human / musical
ensemble / original cast) excludes by design.
justification recorded here. Currently empty: the only pin ever
needed, "Pinkfong" (typed as a brand, which the type gate
excludes by design), became moot when the store's artist entity
behind that credit was identified as Hope Segoine.
A pinned name skips the candidate retrieval and corroboration
steps; its item ID is used directly."""
@@ -143,6 +152,53 @@ SNAPSHOT_FIELDS: Sequence[str] = tuple(
"""The header columns of the Wikidata artist snapshot CSV file."""
@dataclass
class GroupMember:
"""One has-part member of a Wikidata group item."""
qid: str
"""The item ID of the member."""
start_years: list[int] = field(default_factory=list)
"""The years the membership started, from the start-time
qualifiers of the statement."""
end_years: list[int] = field(default_factory=list)
"""The years the membership ended, from the end-time
qualifiers of the statement."""
def performed(self) -> bool:
"""Return whether the member was in the group within the
corpus window.
A membership that ended before the first year of the
corpus window is taken up again when a later start time
says so: Wikidata models a departure and a re-join as
several start times on one membership statement.
:return: False when the membership ended before the first
year of the corpus window and did not start again
afterwards, True otherwise.
"""
if len(self.end_years) == 0:
return True
last_end: int = max(self.end_years)
if last_end >= CORPUS_START_YEAR:
return True
if len(self.start_years) == 0:
return False
return max(self.start_years) > last_end
@dataclass
class MemberClaims:
"""The item-ID claim targets of a Wikidata group member item
that decide the gender of its group."""
instance_of_ids: list[str] = field(default_factory=list)
"""The item IDs of the instance-of targets."""
gender_ids: list[str] = field(default_factory=list)
"""The item IDs of the gender targets."""
@dataclass
class ArtistClaims:
"""The item-ID claim targets and description of a Wikidata
@@ -158,6 +214,8 @@ class ArtistClaims:
"""The item IDs of the country-of-citizenship targets."""
origin_country_ids: list[str] = field(default_factory=list)
"""The item IDs of the country-of-origin targets."""
members: list[GroupMember] = field(default_factory=list)
"""The has-part members, in the statement order."""
description: str = ""
"""The English description of the item, or empty when
absent."""
@@ -220,6 +278,17 @@ class ArtistFetcher:
As a last resort, a name listed in ``PINNED_QIDS`` uses its
pinned item ID directly, skipping every step above.
A resolved item with no gender of its own that is typed as a
group takes the gender of its has-part members: the parts
that left the group before the first year of the corpus
window without re-joining it afterwards and the parts that
are not human are dropped, and the genders of the remaining
members decide -- one shared gender becomes the group's,
differing genders make it ``mixed``, and a member with no
gender of its own, or no member left, leaves the group's
gender unresolved. The basis of a derived gender is recorded
in the note.
"""
def __init__(self) -> None:
@@ -400,6 +469,10 @@ class ArtistFetcher:
def __resolve(self, snapshot: ArtistSnapshot) -> None:
"""Resolve the claims of an artist into the snapshot.
A group with no gender of its own takes the gender
derived from its members, with the basis appended to the
note.
:param snapshot: The snapshot, with the QID set.
:return: None.
:raises OSError: On a non-retryable HTTP or network
@@ -424,6 +497,94 @@ class ArtistFetcher:
labels[x] for x in claims.genre_ids if x in labels)
if len(country_ids) > 0:
snapshot.country = labels.get(country_ids[0], "")
if len(claims.gender_ids) == 0 \
and snapshot.type == ArtistType.GROUP:
self.__derive_gender(snapshot, claims.members)
def __derive_gender(self, snapshot: ArtistSnapshot,
members: Sequence[GroupMember]) -> None:
"""Derive the gender of a group from its members.
Sets the gender to the shared gender label of the
members, or to ``mixed`` when the members do not share
one gender, and appends the basis -- each member and its
gender -- to the note. The gender and the note are left
untouched when a member has no gender of its own, or when
no member is left once the members who left the group
before the corpus window without re-joining it afterwards
and the parts that are not human are dropped.
:param snapshot: The snapshot of the group.
:param members: The has-part members of the group item.
:return: None.
:raises OSError: On a non-retryable HTTP or network
error.
:raises RetryExhausted: When the retries on a
transient error are exhausted.
:raises ValueError: On a JSON decoding error.
"""
qids: list[str] = [x.qid for x in members if x.performed()]
if len(qids) == 0:
return
claims: dict[str, MemberClaims] \
= self.__get_member_claims(qids)
genders: list[tuple[str, str]] = []
qid: str
for qid in qids:
member: MemberClaims = claims.get(qid, MemberClaims())
if HUMAN_QID not in member.instance_of_ids:
continue
if len(member.gender_ids) == 0:
return
genders.append((qid, member.gender_ids[0]))
if len(genders) == 0:
return
labels: dict[str, str] = self.__get_labels(
[x[1] for x in genders], any_language=True)
unique: set[str] = {x[1] for x in genders}
snapshot.gender = labels[genders[0][1]] \
if len(unique) == 1 else MIXED_GENDER
basis: str = "gender derived from members: " + "; ".join(
f"{x} {labels[y]}" for x, y in genders)
snapshot.note = f"{snapshot.note}; {basis}" \
if snapshot.note != "" else basis
def __get_member_claims(self, qids: Sequence[str]) \
-> dict[str, MemberClaims]:
"""Fetch the claims of the group member items in one
batch.
:param qids: The item IDs, duplicates allowed.
:return: The instance-of and gender targets, keyed by the
item ID; the items without claims are left out.
:raises OSError: On a non-retryable HTTP or network
error.
:raises RetryExhausted: When the retries on a
transient error are exhausted.
: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": "claims", "format": "json"})
entities: Any = data.get("entities") \
if isinstance(data, dict) else None
if not isinstance(entities, dict):
return {}
members: dict[str, MemberClaims] = {}
qid: str
for qid in unique:
entity: Any = entities.get(qid)
claims: Any = entity.get("claims") \
if isinstance(entity, dict) else None
if not isinstance(claims, dict):
continue
members[qid] = MemberClaims(
instance_of_ids=self.__targets(claims.get("P31")),
gender_ids=self.__targets(claims.get("P21")))
return members
def __get_claims(self, qid: str) -> ArtistClaims:
"""Fetch the claims and the description of a Wikidata
@@ -431,8 +592,8 @@ class ArtistFetcher:
:param qid: The item ID.
:return: The item-ID targets of the gender, instance-of,
genre, and country properties, and the English
description.
genre, country, and has-part properties, and the
English description.
:raises OSError: On a non-retryable HTTP or network
error.
:raises RetryExhausted: When the retries on a
@@ -457,8 +618,70 @@ class ArtistFetcher:
genre_ids=self.__targets(claims.get("P136")),
country_ids=self.__targets(claims.get("P27")),
origin_country_ids=self.__targets(claims.get("P495")),
members=self.__members(claims.get("P527")),
description=self.__description(entity))
@staticmethod
def __members(statements: Any) -> list[GroupMember]:
"""Extract the members of the has-part statements.
:param statements: The statements of the has-part
property, or None.
:return: The members, in the statement order, each with
the start and end years of its statement.
"""
if not isinstance(statements, list):
return []
members: list[GroupMember] = []
statement: Any
for statement in statements:
if not isinstance(statement, dict):
continue
qids: list[str] = ArtistFetcher.__targets([statement])
if len(qids) == 0:
continue
members.append(GroupMember(
qid=qids[0],
start_years=ArtistFetcher.__years(
statement, "P580"),
end_years=ArtistFetcher.__years(
statement, "P582")))
return members
@staticmethod
def __years(statement: dict[str, Any],
qualifier: str) -> list[int]:
"""Extract the years of a time qualifier of a statement.
:param statement: The statement data.
:param qualifier: The property of the time qualifier.
:return: The years of the qualifiers, in the qualifier
order; the qualifiers of a coarser precision than the
year are read as their year all the same.
"""
qualifiers: Any = statement.get("qualifiers")
snaks: Any = qualifiers.get(qualifier) \
if isinstance(qualifiers, dict) else None
if not isinstance(snaks, list):
return []
years: list[int] = []
snak: Any
for snak in snaks:
if not isinstance(snak, dict):
continue
datavalue: Any = snak.get("datavalue")
if not isinstance(datavalue, dict):
continue
value: Any = datavalue.get("value")
if not isinstance(value, dict) \
or not isinstance(value.get("time"), str):
continue
match: re.Match[str] | None \
= TIME_YEAR_PATTERN.match(value["time"])
if match is not None:
years.append(int(match.group()))
return years
@staticmethod
def __description(entity: Any) -> str:
"""Extract the English description of a Wikidata entity.
@@ -503,13 +726,20 @@ class ArtistFetcher:
ids.append(value["id"])
return ids
def __get_labels(self, qids: Sequence[str]) \
def __get_labels(self, qids: Sequence[str],
any_language: bool = False) \
-> dict[str, str]:
"""Resolve item IDs to their English labels in one batch.
"""Resolve item IDs to their 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.
:param any_language: True to accept the label of another
language for an item without an English label, and to
fall back to the bare item ID for an item without any
label, so that every requested item ID is a key of
the result.
:return: The labels, keyed by the item ID; without
``any_language``, the items without an English label
are left out.
:raises OSError: On a non-retryable HTTP or network
error.
:raises RetryExhausted: When the retries on a
@@ -519,27 +749,51 @@ class ArtistFetcher:
unique: list[str] = list(dict.fromkeys(qids))
if len(unique) == 0:
return {}
data: Any = self.__get_json({
params: dict[str, str] = {
"action": "wbgetentities", "ids": "|".join(unique),
"props": "labels", "languages": "en",
"format": "json"})
"props": "labels", "format": "json"}
if not any_language:
params["languages"] = "en"
data: Any = self.__get_json(params)
entities: Any = data.get("entities") \
if isinstance(data, dict) else None
if not isinstance(entities, dict):
return {}
entities = {}
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"]
label: str = self.__label(
entities.get(qid), any_language)
if label == "" and any_language:
label = qid
if label != "":
labels[qid] = label
return labels
@staticmethod
def __label(entity: Any, any_language: bool) -> str:
"""Extract the label of a Wikidata entity.
:param entity: The entity data, or None.
:param any_language: True to accept the label of another
language when there is no English label.
:return: The label, or the empty string when there is
none to take.
"""
labels: Any = entity.get("labels") \
if isinstance(entity, dict) else None
if not isinstance(labels, dict):
return ""
candidates: list[Any] = [labels.get("en")]
if any_language:
candidates.extend(labels.values())
candidate: Any
for candidate in candidates:
if isinstance(candidate, dict) \
and isinstance(candidate.get("value"), str):
return candidate["value"]
return ""
@staticmethod
def __artist_type(type_ids: Sequence[str],
labels: dict[str, str]) \
@@ -769,20 +1023,31 @@ def append_row(file: TextIO, snapshot: ArtistSnapshot) -> None:
file.flush()
def write_snapshot(file: TextIO) -> None:
def write_snapshot(file: TextIO, names: Container[str]) -> None:
"""Rewrite a snapshot CSV file handle sorted by artist name.
Reads back the current rows and rewrites the header and the
rows ordered by the case-folded artist name, matching the
convention of the derived ``artists.csv``.
The rows are ordered by the case-folded artist name, matching
the convention of the derived ``artists.csv``. An artist
keeps one row only, the last one of the file, so that a
re-fetched artist replaces its earlier row. A row whose name
is not an artist of the store is dropped and reported on the
standard error.
:param file: The open, seekable snapshot CSV file.
:param names: The artist names of the working store.
:return: None.
:raises OSError: When the file cannot be read or written.
"""
kept: dict[str, dict[str, str]] = {}
row: dict[str, str]
for row in read_snapshot_rows(file):
if row["name"] not in names:
print(f"dropped stale row \"{row['name']}\":"
" no such artist in the store", file=sys.stderr)
continue
kept[row["name"]] = row
ordered: list[dict[str, str]] = sorted(
read_snapshot_rows(file),
key=lambda row: row["name"].casefold())
kept.values(), key=lambda x: x["name"].casefold())
file.seek(0)
file.truncate()
writer: csv.DictWriter[str] = csv.DictWriter(
@@ -812,11 +1077,14 @@ def main(argv: list[str] | None = None) -> int:
with open(args.wikidata_csv, "a+", encoding="utf-8",
newline="") as csv_file:
done: set[str] = {x["name"] for x in
read_snapshot_rows(csv_file)}
read_snapshot_rows(csv_file)
if x["gender"] != ""}
ensure_snapshot_header(csv_file)
names: set[str] = set()
artist: Artist
for artist in session.scalars(
sa.select(Artist).order_by(Artist.id)):
names.add(artist.name)
if artist.name in done:
continue
titles: list[str] = read_artist_titles(
@@ -835,7 +1103,7 @@ def main(argv: list[str] | None = None) -> int:
fetched += 1
print(f"artist \"{artist.name}\": {status}",
file=sys.stderr)
write_snapshot(csv_file)
write_snapshot(csv_file, names)
except (OSError, sa.exc.SQLAlchemyError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
tools/tests/test_build_db.py
+62 -3
View File
@@ -583,9 +583,12 @@ class TestBuildDB(unittest.TestCase):
"name,qid,gender,type,genre,country,note\n"
"Adele,Q2831,female,solo,pop,GB,\n"
"Drake,Q33240,male,solo,hip-hop,CA,\n"
"Taylor Swift,Q26876,female,solo,pop,US,\n")
"""The artist snapshot fixture for the performer gender, leaving
"Nobody" and "Nobody Else" without a gender."""
"Taylor Swift,Q26876,female,solo,pop,US,\n"
"Nobody,,,solo,pop,US,\n"
"Nobody Else,,,solo,pop,US,\n")
"""The artist snapshot fixture for the performer gender, giving
"Nobody" and "Nobody Else" a performing type without a
gender."""
def __performer_genders(self) -> dict[str, str | None]:
"""Read the stored performer genders keyed by the titles.
@@ -632,6 +635,62 @@ class TestBuildDB(unittest.TestCase):
("No Gender Song", ""),
("Unknown Song", "")])
NON_PERFORMING_CHART_CSV: str = (
"year,rank,title,artist\n"
"2016,1,Lemonade,"
"Internet Money & Gunna Featuring Don Toliver\n"
"2016,2,Bruno,\"Adassa, Rhenzy Feliz & Encanto Cast\"\n"
"2017,1,Label Song,Internet Money & Encanto Cast\n"
"2017,2,filler,Filler Artist\n")
"""The chart CSV fixture exercising the non-performing credits:
a non-performing credit alongside agreeing performers, one
alongside disagreeing performers, and a song credited to
non-performing artists only."""
NON_PERFORMING_WIKIDATA_CSV: str = (
"name,qid,gender,type,genre,country,note\n"
"Gunna,Q55613105,male,solo,hip-hop,US,\n"
"Don Toliver,Q56513383,male,solo,hip-hop,US,\n"
"Adassa,Q576181,female,solo,pop,US,\n"
"Rhenzy Feliz,Q34344805,male,solo,pop,US,\n"
"Internet Money,Q99691610,,,hip-hop,US,\n"
"Encanto Cast,Q140814124,,,,US,\n")
"""The artist snapshot fixture for the non-performing credits,
leaving "Internet Money" and "Encanto Cast" without a gender and
without a type."""
def __build_non_performing(self) -> dict[str, str | None]:
"""Build the non-performing credit fixture and read back the
performer genders.
:return: The stored performer genders, keyed by the song
titles.
"""
self.__write_chart(self.NON_PERFORMING_CHART_CSV)
self.__write_wikidata(self.NON_PERFORMING_WIKIDATA_CSV)
self.assertEqual(
self.__run_build("--wikidata-csv",
str(self.__wikidata))[0], 0)
return self.__performer_genders()
def test_non_performing_credit_does_not_block(self) -> None:
"""Test that a credited artist without a type does not block
the agreement of the performing artists."""
self.assertEqual(
self.__build_non_performing()["Lemonade"], "male")
def test_non_performing_credit_keeps_mixed(self) -> None:
"""Test that a credited artist without a type leaves a
disagreement among the performing artists as "mixed"."""
self.assertEqual(
self.__build_non_performing()["Bruno"], "mixed")
def test_all_non_performing_credits_unset(self) -> None:
"""Test that a song credited to artists without a type only
leaves the performer gender unset."""
self.assertIsNone(
self.__build_non_performing()["Label Song"])
def test_first_run_on_fresh_store(self) -> None:
"""Test that a build on a fresh store creates the tables."""
self.assertEqual(
tools/tests/test_fetch_artists.py
+330 -10
View File
@@ -29,6 +29,12 @@ class TestFetchArtists(unittest.TestCase):
"name", "qid", "gender", "type", "genre",
"country", "note"]
"""The expected header row of the snapshot CSV file."""
GROUP_QID: str = "Q215380"
"""The item ID of the instance-of target of a group."""
MALE_QID: str = "Q6581097"
"""The item ID of the gender target of a male member."""
FEMALE_QID: str = "Q6581072"
"""The item ID of the gender target of a female member."""
def setUp(self) -> None:
"""Create a temporary capture directory with the store."""
@@ -127,6 +133,69 @@ class TestFetchArtists(unittest.TestCase):
return {"mainsnak": {"snaktype": "value",
"datavalue": {"value": {"id": qid}}}}
@staticmethod
def __part(qid: str, end: str = "",
starts: list[str] | None = None) -> dict[str, Any]:
"""Build a has-part claim statement.
:param qid: The item ID of the member.
:param end: The end time of the membership, as a Wikidata
time value, or the empty string for a membership that
did not end.
:param starts: The start times of the membership, as
Wikidata time values, or None for a membership
without a start time.
:return: The claim statement.
"""
statement: dict[str, Any] \
= TestFetchArtists.__claim(qid)
qualifiers: dict[str, Any] = {}
if starts is not None:
qualifiers["P580"] = [
TestFetchArtists.__time(x) for x in starts]
if end != "":
qualifiers["P582"] = [TestFetchArtists.__time(end)]
if len(qualifiers) > 0:
statement["qualifiers"] = qualifiers
return statement
@staticmethod
def __time(time: str) -> dict[str, Any]:
"""Build a time qualifier snak.
:param time: The time, as a Wikidata time value.
:return: The qualifier snak.
"""
return {"snaktype": "value",
"datavalue": {"value": {"time": time,
"precision": 9}}}
@staticmethod
def __human(gender: str = "") -> dict[str, Any]:
"""Build the claims of a human group member.
:param gender: The item ID of the gender target, or the
empty string for a member without a gender.
:return: The claim statements, keyed by the property.
"""
claims: dict[str, Any] = {
"P31": [TestFetchArtists.__claim("Q5")]}
if gender != "":
claims["P21"] = [TestFetchArtists.__claim(gender)]
return claims
@staticmethod
def __member_claims(members: dict[str, dict[str, Any]]) \
-> dict[str, Any]:
"""Build a member claims response payload.
:param members: The claim statements of each member,
keyed by the member item ID.
:return: The response payload.
"""
return {"entities": {
x: {"claims": y} for x, y in members.items()}}
@staticmethod
def __labels(labels: dict[str, str]) -> dict[str, Any]:
"""Build a label query response payload.
@@ -283,14 +352,18 @@ class TestFetchArtists(unittest.TestCase):
def test_pinned_qid_skips_search(self) -> None:
"""Test that a pinned name short-circuits the search."""
self.__seed(["Pinkfong"])
qid: str = fetch_artists.PINNED_QIDS["Pinkfong"]
self.__seed(["Brandy Brand"])
qid: str = "Q99999999"
claims: dict[str, Any] = self.__claims(
qid, {}, "South Korean children's brand")
qid, {}, "a brand the type gate excludes")
urlopen: mock.Mock
with mock.patch(
"urllib.request.urlopen",
side_effect=[self.__response(claims)]) as urlopen:
with mock.patch.object(
fetch_artists, "PINNED_QIDS",
{"Brandy Brand": qid}), \
mock.patch(
"urllib.request.urlopen",
side_effect=[self.__response(claims)]) \
as urlopen:
status: int = self.__run_fetch()[0]
self.assertEqual(status, 0)
self.assertEqual(urlopen.call_count, 1)
@@ -300,8 +373,8 @@ class TestFetchArtists(unittest.TestCase):
rows: list[list[str]] = self.__read_rows(
self.__snapshot)
self.assertEqual(rows[1], [
"Pinkfong", qid, "", "", "",
"", "South Korean children's brand"])
"Brandy Brand", qid, "", "", "",
"", "a brand the type gate excludes"])
def test_stage1_performer_intersection(self) -> None:
"""Test the multi-candidate resolution via a performer."""
@@ -383,6 +456,193 @@ class TestFetchArtists(unittest.TestCase):
self.assertEqual(rows[1], [
"Case", "Q2", "", "", "", "", "Rescued by casefold"])
def __fetch_group(self, parts: list[dict[str, Any]],
members: dict[str, dict[str, Any]],
gender_labels: dict[str, Any] | None) \
-> tuple[list[str], mock.Mock]:
"""Run the fetcher on a store holding one group artist.
:param parts: The has-part claim statements of the group
item.
:param members: The claim statements of each member item,
keyed by the member item ID, or empty when no member
item is expected to be queried.
:param gender_labels: The label response payload of the
gender items, or None when no label is expected to be
queried.
:return: A tuple of the snapshot row of the group and the
``urlopen`` mock.
"""
self.__seed(["Boyz"])
responses: list[Any] = [
self.__response(self.__sparql(
[{"item": self.__uri("Q10")}])),
self.__response(self.__claims(
"Q10",
{"P31": [self.__claim(self.GROUP_QID)],
"P527": parts},
"American boy band")),
self.__response(self.__labels(
{self.GROUP_QID: "musical group"}))]
if len(members) > 0:
responses.append(self.__response(
self.__member_claims(members)))
if gender_labels is not None:
responses.append(self.__response(gender_labels))
urlopen: mock.Mock
with mock.patch("urllib.request.urlopen",
side_effect=responses) as urlopen:
status: int = self.__run_fetch()[0]
self.assertEqual(status, 0)
self.assertEqual(urlopen.call_count, len(responses))
rows: list[list[str]] = self.__read_rows(self.__snapshot)
self.assertEqual(len(rows), 2)
return rows[1], urlopen
def test_group_gender_from_members(self) -> None:
"""Test a group taking the shared gender of its
members."""
row: list[str] = self.__fetch_group(
[self.__part("Q11"), self.__part("Q12")],
{"Q11": self.__human(self.MALE_QID),
"Q12": self.__human(self.MALE_QID)},
self.__labels({self.MALE_QID: "male"}))[0]
self.assertEqual(row, [
"Boyz", "Q10", "male", "group", "", "",
"American boy band; gender derived from members:"
" Q11 male; Q12 male"])
def test_group_gender_mixed(self) -> None:
"""Test a group whose members do not share one gender."""
row: list[str] = self.__fetch_group(
[self.__part("Q11"), self.__part("Q12")],
{"Q11": self.__human(self.MALE_QID),
"Q12": self.__human(self.FEMALE_QID)},
self.__labels({self.MALE_QID: "male",
self.FEMALE_QID: "female"}))[0]
self.assertEqual(row, [
"Boyz", "Q10", "mixed", "group", "", "",
"American boy band; gender derived from members:"
" Q11 male; Q12 female"])
def test_group_member_without_gender(self) -> None:
"""Test that a member without a gender of its own leaves
the gender of its group unresolved."""
row: list[str] = self.__fetch_group(
[self.__part("Q11"), self.__part("Q12")],
{"Q11": self.__human(self.MALE_QID),
"Q12": self.__human()},
None)[0]
self.assertEqual(row, [
"Boyz", "Q10", "", "group", "", "",
"American boy band"])
def test_group_non_human_part_ignored(self) -> None:
"""Test that a part that is not human does not count."""
row: list[str] = self.__fetch_group(
[self.__part("Q11"), self.__part("Q13")],
{"Q11": self.__human(self.MALE_QID),
"Q13": {"P31": [self.__claim("Q2088357")]}},
self.__labels({self.MALE_QID: "male"}))[0]
self.assertEqual(row, [
"Boyz", "Q10", "male", "group", "", "",
"American boy band; gender derived from members:"
" Q11 male"])
def test_group_member_left_before_corpus(self) -> None:
"""Test that a member who left before the corpus window
is neither queried nor counted."""
row: list[str]
urlopen: mock.Mock
row, urlopen = self.__fetch_group(
[self.__part("Q11", "+2015-06-01T00:00:00Z"),
self.__part("Q12")],
{"Q12": self.__human(self.MALE_QID)},
self.__labels({self.MALE_QID: "male"}))
self.assertEqual(row, [
"Boyz", "Q10", "male", "group", "", "",
"American boy band; gender derived from members:"
" Q12 male"])
url: str = urlopen.call_args_list[3][0][0].full_url
self.assertIn("Q12", url)
self.assertNotIn("Q11", url)
def test_group_member_left_within_corpus(self) -> None:
"""Test that a member who left within the corpus window
still counts."""
row: list[str] = self.__fetch_group(
[self.__part("Q11", "+2016-03-01T00:00:00Z"),
self.__part("Q12")],
{"Q11": self.__human(self.FEMALE_QID),
"Q12": self.__human(self.MALE_QID)},
self.__labels({self.MALE_QID: "male",
self.FEMALE_QID: "female"}))[0]
self.assertEqual(row, [
"Boyz", "Q10", "mixed", "group", "", "",
"American boy band; gender derived from members:"
" Q11 female; Q12 male"])
def test_group_member_rejoined_after_leaving(self) -> None:
"""Test that a member who left before the corpus window
and re-joined after it still counts."""
row: list[str] = self.__fetch_group(
[self.__part("Q11", "+2013-01-01T00:00:00Z",
["+2005-01-01T00:00:00Z",
"+2019-01-01T00:00:00Z"]),
self.__part("Q12")],
{"Q11": self.__human(self.FEMALE_QID),
"Q12": self.__human(self.MALE_QID)},
self.__labels({self.MALE_QID: "male",
self.FEMALE_QID: "female"}))[0]
self.assertEqual(row, [
"Boyz", "Q10", "mixed", "group", "", "",
"American boy band; gender derived from members:"
" Q11 female; Q12 male"])
def test_group_member_left_for_good(self) -> None:
"""Test that a member who left before the corpus window
and did not re-join is excluded."""
row: list[str]
urlopen: mock.Mock
row, urlopen = self.__fetch_group(
[self.__part("Q11", "+2013-01-01T00:00:00Z",
["+2005-01-01T00:00:00Z"]),
self.__part("Q12")],
{"Q12": self.__human(self.MALE_QID)},
self.__labels({self.MALE_QID: "male"}))
self.assertEqual(row, [
"Boyz", "Q10", "male", "group", "", "",
"American boy band; gender derived from members:"
" Q12 male"])
url: str = urlopen.call_args_list[3][0][0].full_url
self.assertIn("Q12", url)
self.assertNotIn("Q11", url)
def test_group_gender_label_without_english(self) -> None:
"""Test the gender label of another language being taken
when there is no English one."""
row: list[str] = self.__fetch_group(
[self.__part("Q11")],
{"Q11": self.__human(self.MALE_QID)},
{"entities": {self.MALE_QID: {
"labels": {"ja": {"value": "男性"}}}}})[0]
self.assertEqual(row, [
"Boyz", "Q10", "男性", "group", "", "",
"American boy band; gender derived from members:"
" Q11 男性"])
def test_group_gender_label_missing(self) -> None:
"""Test the bare item ID being taken as the gender label
when the item has no label at all."""
row: list[str] = self.__fetch_group(
[self.__part("Q11")],
{"Q11": self.__human(self.MALE_QID)},
{"entities": {}})[0]
self.assertEqual(row, [
"Boyz", "Q10", self.MALE_QID, "group", "", "",
"American boy band; gender derived from members:"
f" Q11 {self.MALE_QID}"])
def test_unresolved_continues_to_next_artist(self) -> None:
"""Test that an unresolved artist does not stop the run."""
self.__seed(["Ambiguous", "Nobody"])
@@ -527,15 +787,75 @@ class TestFetchArtists(unittest.TestCase):
"Nobody", "", "", "", "", "", "not found"])
self.__assert_summary(stderr, 0, 1)
def test_blank_gender_row_refetched(self) -> None:
"""Test that a row with a blank gender is re-fetched and
replaced, not duplicated."""
self.__seed(["Adele"])
snapshot: Path = self.__snapshot
with open(snapshot, "w", encoding="utf-8",
newline="") as file:
writer: Any = csv.writer(file)
writer.writerow(self.HEADER)
writer.writerow([
"Adele", "Q1", "", "solo", "", "", "not found"])
candidates: dict[str, Any] = self.__sparql(
[{"item": self.__uri("Q1")}])
claims: dict[str, Any] = self.__claims(
"Q1", {"P21": [self.__claim("Q2")],
"P31": [self.__claim("Q5")]},
"English singer")
labels: dict[str, Any] = self.__labels(
{"Q2": "female", "Q5": "human"})
urlopen: mock.Mock
with mock.patch(
"urllib.request.urlopen",
side_effect=[self.__response(candidates),
self.__response(claims),
self.__response(labels)]) as urlopen:
status: int = self.__run_fetch()[0]
self.assertEqual(status, 0)
self.assertEqual(urlopen.call_count, 3)
rows: list[list[str]] = self.__read_rows(snapshot)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[1], [
"Adele", "Q1", "female", "solo", "", "",
"English singer"])
def test_stale_row_dropped(self) -> None:
"""Test that a row matching no artist of the store is
dropped and reported."""
self.__seed(["Amy"])
snapshot: Path = self.__snapshot
amy_row: list[str] = [
"Amy", "Q1", "female", "solo", "", "", "English singer"]
with open(snapshot, "w", encoding="utf-8",
newline="") as file:
writer: Any = csv.writer(file)
writer.writerow(self.HEADER)
writer.writerow(amy_row)
writer.writerow([
"Pinkfong", "Q2", "male", "solo", "", "", "a brand"])
urlopen: mock.Mock
with mock.patch("urllib.request.urlopen") as urlopen:
status: int
stderr: str
status, stderr = self.__run_fetch()
self.assertEqual(status, 0)
urlopen.assert_not_called()
rows: list[list[str]] = self.__read_rows(snapshot)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[1], amy_row)
self.assertIn("Pinkfong", stderr)
def test_topup_inserts_sorted_position(self) -> None:
"""Test that a top-up inserts the new row in its sorted
position, not appended at the end of the file."""
self.__seed(["Amy", "Mia", "Zed"])
snapshot: Path = self.__snapshot
amy_row: list[str] = [
"Amy", "Q1", "", "", "", "", "not found"]
"Amy", "Q1", "female", "solo", "", "", "a singer"]
zed_row: list[str] = [
"Zed", "Q2", "", "", "", "", "not found"]
"Zed", "Q2", "male", "solo", "", "", "a singer"]
with open(snapshot, "w", encoding="utf-8",
newline="") as file:
writer: Any = csv.writer(file)