#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""This is **ALIS** (reads *Alice*):
suggest images for Wikipedia **articles** that don't have one.
Inputs come from Wikimedia Foundation's `Analytics Data Lake <https://wikitech.wikimedia.org/wiki/Analytics/Data_Lake>`_:
- tables from the
`raw MediaWiki <https://wikitech.wikimedia.org/wiki/Analytics/Data_Lake/Edits#Raw_Mediawiki_data>`_ database
- `wmf_raw.mediawiki_page <https://www.mediawiki.org/wiki/Manual:Page_table>`_
- `wmf_raw.mediawiki_imagelinks <https://www.mediawiki.org/wiki/Manual:Imagelinks_table>`_
- `wmf_raw.mediawiki_private_linktarget <https://www.mediawiki.org/wiki/Manual:Linktarget_table>`_
- `wmf_raw.mediawiki_revision <https://www.mediawiki.org/wiki/Manual:Revision_table>`_
- `Wikidata item page links <https://wikitech.wikimedia.org/wiki/Analytics/Data_Lake/Content/Wikidata_item_page_link>`_
- ``mediawiki_image_suggestions_feedback`` table from
`event_sanitized <https://wikitech.wikimedia.org/wiki/Analytics/Data_Lake/Events>`_
High-level steps:
- 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>`_
- gather lead images of Wikipedia articles
- gather Commons `depicts statements <https://commons.wikimedia.org/wiki/COM:DEPICTS>`_ with
`depicts <https://www.wikidata.org/wiki/Property:P180>`_,
`is digital representation of <https://www.wikidata.org/wiki/Property:P6243>`_
Wikidata properties
- compute the confidence score depending on the sources above
- collect Wikipedia articles that don't have an image or are suitable for getting suggestions
- filter out irrelevant suggestion candidates, typically placeholders and overused images.
See :func:`image_suggestions.unillustratable.get_images_in_placeholder_categories`
and :func:`image_suggestions.unillustratable.get_overused_images` respectively
- filter out suggestions already reviewed by users
Output :class:`pyspark.sql.Row` example:
.. code:: py
Row(
page_id=9696852,
id='ddd3bad8-327a-11ee-8991-f4e9d4472fd0',
image='Gamez,_Cónsul_-_2018_Junior_Worlds_-_5.jpg',
origin_wiki='commonswiki',
confidence=96,
found_on=['ruwiki'],
kind=['istype-commons-category', 'istype-lead-image'],
page_rev=132612416,
section_heading=None,
section_index=None,
page_qid='Q21660678',
snapshot='2023-07-24',
wiki='itwiki',
)
More `documentation <https://www.mediawiki.org/wiki/Structured_Data_Across_Wikimedia/Image_Suggestions/Data_Pipeline#How_it_works>`_
lives in MediaWiki.
"""
import argparse
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F
import image_suggestions.topic_slis
from image_suggestions import (
entity_images,
overused_images,
queries,
shared,
unillustratable,
)
#: Amount of suggestions per Wikidata QID
LIMIT_PER_QID = 20
[docs]
def load_local_images(
spark: SparkSession, short_snapshot: str
) -> DataFrame: # pragma: no cover
"""Load locally stored images through the :const:`image_suggestions.queries.local_images` Data Lake query.
:param spark: an active Spark session
:param short_snapshot: a ``YYYY-MM`` date
:return: the dataframe of wikis, file page IDs, and file names
"""
return spark.sql(queries.local_images.format(short_snapshot))
[docs]
def load_suggestions_with_feedback(
spark: SparkSession,
) -> DataFrame: # pragma: no cover
"""Load image suggestions that were reviewed by users.
:param spark: an active Spark session
:return: the dataframe of wikis, page IDs, and image file names
"""
return spark.sql(queries.suggestions_with_feedback)
def get_unillustrated_articles(spark: SparkSession, short_snapshot: str) -> DataFrame:
commons_images = shared.load_commons_images(spark, short_snapshot)
overused_images = image_suggestions.overused_images.get(spark, short_snapshot)
local_images = load_local_images(spark, short_snapshot)
imagelinks = shared.load_imagelinks(spark, short_snapshot)
illustrated_articles = (
imagelinks.join(
# Ignore images used so widely that they are likely icons
overused_images,
on=[
imagelinks.wiki_db == overused_images.wiki_db,
imagelinks.image_title == overused_images.page_title,
],
how='left_anti',
)
.join(
commons_images,
on=[imagelinks.image_title == commons_images.page_title],
how='inner',
)
.select(imagelinks.wiki_db, 'article_id')
.union(
imagelinks.join(
local_images,
on=[
imagelinks.image_title == local_images.page_title,
imagelinks.wiki_db == local_images.wiki_db,
],
how='inner',
).select(imagelinks.wiki_db, 'article_id')
)
)
non_commons_main_pages = shared.load_non_commons_main_pages(spark, short_snapshot)
return non_commons_main_pages.join(
illustrated_articles,
on=[
non_commons_main_pages.page_id == illustrated_articles.article_id,
non_commons_main_pages.wiki_db == illustrated_articles.wiki_db,
],
how='left_anti',
).select('wiki_db', 'page_id', 'page_title')
[docs]
def get_illustratable_articles(
spark: SparkSession, hive_db: str, snapshot: str
) -> DataFrame:
"""Collect Wikipedia articles that are suitable candidates for image suggestions.
A candidate has either no images or its images are used so widely
across Wikimedia projects that they are probably icons or placeholders.
:param spark: an active Spark session
:param hive_db: a Data Lake's `Hive <https://hive.apache.org/>`_ database name
:param snapshot: a ``YYYY-MM-DD`` date
:return: the dataframe of wikis, page IDs, page titles, and page QIDs
"""
short_snapshot = shared.get_monthly_snapshot(snapshot)
unillustrated = get_unillustrated_articles(spark, short_snapshot)
wikidata_item_page_links = shared.load_wikidata_item_page_links(
spark, hive_db, snapshot
)
non_illustratable_item_ids = unillustratable.get_non_illustratable_item_ids(
spark, hive_db, snapshot
)
return (
unillustrated.join(wikidata_item_page_links, on=['wiki_db', 'page_title'])
.join(
non_illustratable_item_ids,
on=[wikidata_item_page_links.item_id == non_illustratable_item_ids.item_id],
how='left_anti',
)
.select('wiki_db', 'page_id', 'page_title', 'item_id')
)
def get_illustratable_articles_with_revisions(
spark: SparkSession, hive_db: str, snapshot: str
) -> DataFrame:
short_snapshot = shared.get_monthly_snapshot(snapshot)
illustratable_articles = get_illustratable_articles(spark, hive_db, snapshot)
latest_revisions = shared.load_latest_revisions(spark, short_snapshot)
return illustratable_articles.join(
latest_revisions, on=['wiki_db', 'page_id'], how='inner'
).select('wiki_db', 'page_id', 'page_title', 'item_id', 'rev_id')
[docs]
def generate_suggestions(
spark: SparkSession, hive_db: str, snapshot: str, limit_per_qid: int, coalesce: int
) -> DataFrame:
"""Gather the full ALIS dataset.
:param spark: an active Spark session
:param hive_db: a Data Lake's `Hive <https://hive.apache.org/>`_ database name
:param snapshot: a ``YYYY-MM-DD`` date
: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
:param coalesce: an integer to control the amount of files per output partition.
A higher value implies more files but a faster and lighter execution
:return: the ALIS dataframe
"""
dataset_id = shared.create_dataset_id()
suggestions = entity_images.get(
spark, hive_db, snapshot, limit_per_qid=limit_per_qid
)
articles = get_illustratable_articles_with_revisions(spark, hive_db, snapshot)
suggestions_with_feedback = load_suggestions_with_feedback(spark)
images_in_placeholder_categories = (
unillustratable.get_images_in_placeholder_categories(spark)
)
images_with_too_many_links = overused_images.get(spark, snapshot)
suggestions_full = (
articles.join(suggestions, on=['item_id'], how='inner')
# exclude suggestions with unwanted extensions
.where(
suggestions.page_title.rlike(unillustratable.get_allowed_suffixes_regex())
)
# exclude suggestions with unwanted substrings in title
.where(
suggestions.page_title.rlike(
unillustratable.get_disallowed_substrings_regex()
)
== False
)
# exclude suggestions categorized as placeholders
.join(
images_in_placeholder_categories,
on=[suggestions.page_title == images_in_placeholder_categories.page_title],
how='left_anti',
)
# exclude suggestions used so frequently they're likely icons
.join(
images_with_too_many_links,
on=[
images_with_too_many_links.wiki_db == articles.wiki_db,
images_with_too_many_links.page_title == suggestions.page_title,
],
how='left_anti',
)
# exclude suggestions that have already had feedback
.join(
suggestions_with_feedback,
on=[
articles.wiki_db == suggestions_with_feedback.wiki,
articles.page_id == suggestions_with_feedback.page_id,
suggestions.page_title == suggestions_with_feedback.filename,
],
how='left_anti',
)
.withColumn(shared.DATASET_ID_COLNAME, F.lit(dataset_id))
.withColumn(shared.ORIGIN_WIKI_COLNAME, F.lit(shared.COMMONSWIKI_VALUE))
.select(
articles.wiki_db.alias('wiki'),
articles.page_id,
articles.page_title,
articles.item_id,
shared.DATASET_ID_COLNAME,
suggestions.page_title.alias('image'),
shared.ORIGIN_WIKI_COLNAME,
suggestions.confidence,
suggestions.found_on,
suggestions.kind,
articles.rev_id.alias('page_rev'),
)
)
suggestions_clean = prepare_for_saving(suggestions_full)
shared.save_suggestions_and_cache(
spark, suggestions_clean, hive_db, snapshot, coalesce
)
# Returning this to make it testable
return suggestions_clean
def prepare_for_saving(suggestions: DataFrame) -> DataFrame:
return shared.add_null_missing_columns(
suggestions.withColumnRenamed('page_title', 'title')
)
def parse_args() -> argparse.Namespace: # pragma: no cover
description = 'Generate image suggestions for candidate Wikipedia articles'
parser = shared.build_base_arg_parser(description)
parser.add_argument(
'-l',
'--limit-per-qid',
metavar='N',
default=LIMIT_PER_QID,
help=(
'Limit the amount of suggestions per Wikidata QID. '
f'Default: {LIMIT_PER_QID}'
),
)
return parser.parse_args()
def main(args: argparse.Namespace, spark: SparkSession) -> None:
hive_db = args.hive_db
snapshot = args.snapshot
coalesce = args.coalesce
limit_per_qid = args.limit_per_qid
generate_suggestions(spark, hive_db, snapshot, limit_per_qid, coalesce)
if __name__ == '__main__':
args = parse_args()
spark = shared.build_spark_session() # Pass 'dev' to switch environments
main(args, spark)
spark.stop()