#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Collect images connected to the following sources:
- Wikidata `image <https://www.wikidata.org/wiki/Property:P18>`_ property
- Wikidata `Commons category <https://www.wikidata.org/wiki/Property:P373>`_ property
- Wikipedia article lead images
- Commons `depicts <https://commons.wikimedia.org/wiki/COM:DEPICTS>`_ statements
"""
from pyspark.sql import DataFrame, SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, LongType, StringType, StructField, StructType
from image_suggestions import queries, shared
[docs]
def get_wikidata(spark: SparkSession, hive_db: str, weekly_snapshot: str) -> DataFrame:
"""Gather `image <https://www.wikidata.org/wiki/Property:P18>`_ and
`Commons category <https://www.wikidata.org/wiki/Property:P373>`_ Wikidata
`claims <https://www.wikidata.org/wiki/WD:GLOSS#Claim>`_.
This function invokes :func:`image_suggestions.shared.load_wikidata`
and aggregates claims by image.
:param spark: an active Spark session
:param hive_db: a Data Lake's `Hive <https://hive.apache.org/>`_ database name
:param weekly_snapshot: a ``YYYY-MM-DD`` date
:return: the dataframe of:
- item_id (string) - Wikidata `QID <https://www.wikidata.org/wiki/WD:GLOSS#QID>`_
- page_id (bigint) - Commons image page ID
- tag (array<string>) - ``image.linked.from.wikidata.p18`` and/or
``image.linked.from.wikidata.p373`` tags
"""
return (
shared.load_wikidata(spark, hive_db, weekly_snapshot)
# One entry may show up more than once, for multiple tags
.groupBy('item_id', 'page_id')
.agg(F.array_distinct(F.collect_list('tag')).alias('tag'))
.select('item_id', 'page_id', 'tag')
)
[docs]
def get_depicts(spark: SparkSession, hive_db: str, weekly_snapshot: str) -> DataFrame:
"""Gather Commons depicts statements.
:param spark: an active Spark session
:param hive_db: a Data Lake's `Hive <https://hive.apache.org/>`_ database name
:param weekly_snapshot: a ``YYYY-MM-DD`` date
:return: the dataframe of:
- item_id (string) - Wikidata `QID <https://www.wikidata.org/wiki/WD:GLOSS#QID>`_
- page_id (bigint) - Commons image page ID
"""
return spark.sql(
queries.commons_pages_with_depicts.format(hive_db, weekly_snapshot)
)
[docs]
def get(
spark: SparkSession,
hive_db: str,
weekly_snapshot: str,
include_wikidata: bool = True,
include_lead_images: bool = True,
include_sdc: bool = True,
limit_per_qid: int = 0,
) -> DataFrame:
"""Aggregate all sources of image connections.
Compute an image suggestion confidence score between 0 and 100 based on the sources:
if an image has one source, it will inherit its score.
Otherwise, it will be a combined *or* probability.
For instance, Commons categories and depicts statements have a score of
``80`` and ``70`` respectively. If an image is connected to both of them,
then the final score will be:
::
100 x ( 1 - (1 - 0.8) x (1 - 0.7) ) = 97
More details
`here <https://www.mediawiki.org/wiki/Structured_Data_Across_Wikimedia/Image_Suggestions/Data_Pipeline#Confidence_score>`_.
:param spark: an active Spark session
:param hive_db: a Data Lake's `Hive <https://hive.apache.org/>`_ database name
:param weekly_snapshot: a ``YYYY-MM-DD`` date
:param include_wikidata: whether to include Wikidata claims
:param include_lead_images: whether to include Wikipedia article lead images
:param include_sdc: whether to include Commons depicts statements
:param limit_per_qid: an integer that limits the amount of suggestions per Wikidata QID.
If ``> 0``, suggestions are ordered by confidence. ``0`` stands for no limit
:return: the dataframe of:
- item_id (string) - Wikidata QID
- page_title (string) - Commons image page title, in original case and underscored
- found_on (array<string>) - wikis where the image was found
- kind (array<string>) - sources of image connections. Values can be
``istype-depicts``, ``istype-wikidata-image``,
``istype-commons-category``, and/or ``istype-lead-image``
- confidence (int) - suggestion confidence score
"""
monthly_snapshot = shared.get_monthly_snapshot(weekly_snapshot)
commons_images = shared.load_commons_images(spark, monthly_snapshot)
wikidata = (
spark.createDataFrame(
[],
StructType(
[
StructField('item_id', StringType(), True),
StructField('page_id', LongType(), True),
StructField('tag', ArrayType(StringType()), True),
]
),
)
if not include_wikidata
else get_wikidata(spark, hive_db, weekly_snapshot)
)
lead_images = (
spark.createDataFrame(
[],
StructType(
[
StructField('item_id', StringType(), True),
StructField('page_id', LongType(), True),
StructField('found_on', ArrayType(StringType()), True),
]
),
)
if not include_lead_images
else shared.load_lead_images(spark, hive_db, weekly_snapshot)
)
depicts = (
spark.createDataFrame(
[],
StructType(
[
StructField('item_id', StringType(), True),
StructField('page_id', LongType(), True),
]
),
)
if not include_sdc
else get_depicts(spark, hive_db, weekly_snapshot)
).withColumn('depicts', F.lit(1))
all_sources = (
wikidata.join(lead_images, on=['item_id', 'page_id'], how='outer')
.join(depicts, on=['item_id', 'page_id'], how='outer')
.join(commons_images, on=['page_id'], how='inner')
.select(
'item_id',
'page_id',
'page_title',
lead_images.found_on.alias(shared.FOUND_ON_COLNAME),
# Compile array to describe the sources where the suggestion was found
F.array_except(
F.array(
F.when(
F.array_contains(
wikidata.tag, 'image.linked.from.wikidata.p18'
),
F.lit('istype-wikidata-image'),
),
F.when(
F.array_contains(
wikidata.tag, 'image.linked.from.wikidata.p373'
),
F.lit('istype-commons-category'),
),
F.when(
lead_images.found_on.isNotNull(), F.lit('istype-lead-image')
),
F.when(depicts.depicts.isNotNull(), F.lit('istype-depicts')),
),
F.array(F.lit(None)),
).alias(shared.KIND_COLNAME),
# Assign a confidence score (within a 0-100 range) based on the source of the suggestion;
# if it matches just 1 source (e.g. image.linked.from.wikidata.p18), it'll have that source's
# confidence (e.g. 100 * (1 - (1 - 0.9)) = 90); if it matches multiple (e.g. image.linked.from.wikidata.p373
# and SDC), it's going to be a combined OR probability (e.g. 100 * (1 - (1 - 0.9) * (1 - 0.7)) = 97)
F.round(
F.lit(100)
* (
F.lit(1)
- (
F.lit(1)
- F.when(
F.array_contains(
wikidata.tag, 'image.linked.from.wikidata.p18'
),
F.lit(0.9),
).otherwise(0)
)
* (
F.lit(1)
- F.when(
F.array_contains(
wikidata.tag, 'image.linked.from.wikidata.p373'
),
F.lit(0.8),
).otherwise(0)
)
* (
F.lit(1)
- F.when(
lead_images.found_on.isNotNull(), F.lit(0.8)
).otherwise(0)
)
* (
F.lit(1)
- F.when(depicts.depicts.isNotNull(), F.lit(0.7)).otherwise(0)
)
)
).alias(shared.CONFIDENCE_COLNAME),
)
)
if limit_per_qid > 0:
window = Window.partitionBy('item_id').orderBy(
all_sources.confidence.desc(), all_sources.page_id
)
all_sources = (
all_sources.withColumn('row_num', F.row_number().over(window))
.where(f'row_num <= {limit_per_qid}')
.drop('row_num')
)
return all_sources.drop('page_id')