103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
import logging
|
|
import socket
|
|
|
|
import pytest
|
|
from django.db.models import Q
|
|
from wikibaseintegrator import WikibaseIntegrator, wbi_helpers
|
|
from wikibaseintegrator.wbi_config import config as wbi_config
|
|
|
|
from funkwhale_api.music.models import Artist
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def enable_network_calls():
|
|
socket.socket = socket.create_connection
|
|
|
|
|
|
WIKIDATA_QUERY = """SELECT DISTINCT ?item ?itemLabel WHERE {
|
|
SERVICE wikibase:label { bd:serviceParam wikibase:language "[AUTO_LANGUAGE]". }
|
|
|
|
?item wdt:P31 wd:Q215380.
|
|
# Match items with the relevant properties and ensure they have references
|
|
{
|
|
VALUES ?genre {
|
|
wd:Q533914 # NSBM
|
|
wd:Q224694 # whit power music
|
|
wd:Q113084468 # nazi rock
|
|
wd:Q121411631 # neonazi music
|
|
wd:Q1547998 # rock identitaire francai
|
|
wd:Q602498 # nazi punk
|
|
wd:Q3328582 # italian right wing alternative
|
|
wd:Q828181 # rock against communism
|
|
}
|
|
|
|
?item p:P136 ?statement.
|
|
?statement ps:P136 ?genre.
|
|
|
|
|
|
# Ensure that these statements have references
|
|
FILTER EXISTS { ?statement prov:wasDerivedFrom ?reference. }
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
def get_far_right_artists():
|
|
from funkwhale_api.music.tasks import get_best_candidate_or_create
|
|
|
|
wbi_config[
|
|
"USER_AGENT"
|
|
] = "Funkwhale_far_righ_artist/1.0 (https://docs.funkwhale.audio/specs/far-right-filter/index.html)"
|
|
artists = []
|
|
wbi = WikibaseIntegrator()
|
|
|
|
results = wbi_helpers.execute_sparql_query(WIKIDATA_QUERY)
|
|
for result in results["results"]["bindings"]:
|
|
item_id = result["itemLabel"]["value"]
|
|
wkd_artist = wbi.item.get(item_id).get_json()
|
|
artist_name = (
|
|
wkd_artist["labels"].get("mul", {}).get("value")
|
|
or wkd_artist["labels"].get("en", {}).get("value")
|
|
or next(iter(wkd_artist["labels"].values()), False)
|
|
)
|
|
|
|
if not artist_name:
|
|
logger.info(f"Artist {item_id} has no label, skipping")
|
|
continue
|
|
|
|
query = Q(name=artist_name)
|
|
|
|
if wkd_artist["claims"].get("P434", False):
|
|
artist_mbid = wkd_artist["claims"]["P434"][0]["mainsnak"]["datavalue"][
|
|
"value"
|
|
]
|
|
query = query & Q(mbid=artist_mbid)
|
|
else:
|
|
logger.warning(
|
|
f"Artist {artist_name} from https://www.wikidata.org/wiki/{item_id} has no MBID. Skipping creation."
|
|
)
|
|
continue
|
|
|
|
default = {"name": artist_name, "mbid": artist_mbid, "far_right": item_id}
|
|
|
|
artist, created = get_best_candidate_or_create(
|
|
Artist, query, default, sort_fields=["name", "mbid"]
|
|
)
|
|
|
|
if not created:
|
|
artist.far_right = item_id
|
|
artist.save()
|
|
|
|
artists.append(artist)
|
|
|
|
logger.info(f"Found {len(artists)} far right artists : {[a.name for a in artists]}")
|
|
|
|
# Remove far right flag from artists that are not in the list
|
|
Artist.objects.filter(far_right__isnull=False).exclude(
|
|
id__in=[a.id for a in artists]
|
|
).update(far_right=None)
|
|
|
|
return artists
|