wikimedica-disease-search/data/pageviews/pageviews.py

import calendar
import collections
from datetime import datetime
import logging
import numpy
from matplotlib import pyplot
import requests
from scipy import stats
import sys
import mwclient

# Chemin racine pour les API Wikimedia
wikimedia_base_path = 'https://wikimedia.org/api/rest_v1'

# Patron d’accès à l’API pageviews de Wikimedia
wikimedia_pageviews_path = '/' + '/'.join([
    'metrics', 'pageviews', 'per-article', '{project}',
    '{access}', '{agent}', '{article}', '{granularity}',
    '{start}', '{end}'
])

# Format de dates utilisée pour l’API Wikimedia
wikimedia_date_format = '%Y%m%d'

# Date de première disponibilité des pageviews sur l’API Wikimedia
wikimedia_pageviews_start = datetime(2015, 7, 1)

# Objet pour afficher le journal d’exécution
logger = logging.getLogger('pageviews')

# Tableau contenant tous les jours de l’année de 1 à 365
year_all_days = numpy.arange(1, 366)


def year_date_distance(a, b):
    """
    Calcule la distance entre deux jours de l’année.

    :example: year_date_distance(10, 360) == 15
    :example: year_date_distance(numpy.array([10, 182, 355]), 182)
              == [172, 0, 173]
    :param a: Première valeur (peut être un tableau numpy).
    :param b: Seconde valeur (peut être un tableau numpy).
    :return: Valeur de la distance.
    """
    return numpy.stack((
        numpy.mod(a - b, len(year_all_days)),
        numpy.mod(b - a, len(year_all_days))
    )).min(axis=0)


def year_smooth_gaussian(data, scale):
    """
    Lisse un tableau de valeurs contenant une valeur par jour de l’année en
    utilisant un noyau gaussien.

    À la bordure de fin ou de début d’année, le lissage est fait avec l’autre
    morceau de l’année.

    :param data: Données à lisser.
    :param scale: Variance du noyau gaussien.
    :return: Données lissées.
    """
    ref_pdf = stats.norm.pdf(year_date_distance(year_all_days, 1), scale=scale)
    pdf_matrix = numpy.stack([
        numpy.roll(ref_pdf, day - 1)
        for day in year_all_days
    ])
    return pdf_matrix.dot(data)


def wikimedia_page_views(site, article):
    """
    Obtient le nombre de visites sur une page Wikipédia par jour.

    :param site: Site Wikipédia ciblé.
    :param article: Article ciblé dans le site.
    :return: Compteur associant chaque jour à son nombre de visites.
    """
    # Soumet une requête à l’API REST pour obtenir les vues de l’article
    res = requests.get(wikimedia_base_path + wikimedia_pageviews_path.format(
        project=site.host,
        article=article,
        access='all-access',
        agent='user',
        granularity='daily',
        start=wikimedia_pageviews_start.strftime(wikimedia_date_format),
        end=datetime.today().strftime(wikimedia_date_format)
    ))

    data = res.json()

    # Vérifie que la réponse reçue indique un succès
    if res.status_code != 200:
        if 'detail' in data:
            detail = data['detail']
            message = ', '.join(detail) if type(detail) == list else detail
            raise Exception(message)
        else:
            raise Exception('Erreur {}'.format(res.status_code))

    # Construit le dictionnaire résultant
    return collections.Counter(dict(
        (record['timestamp'][:8], record['views'])
        for record in data['items']
    ))


def wikimedia_article_canonical_name(site, article):
    """
    Obtient le nom canonique d’un article après avoir suivi les redirections.

    :param site: Objet empaquetant l’API du site Wikipédia ciblé.
    :param article: Article ciblé dans le site.
    :return: Nom canonique de l’article.
    """
    original_page = site.pages[article]

    if not original_page.exists:
        raise Exception(
            'Article « {} » inexistant sur {}'
            .format(article, site.host)
        )

    return original_page.resolve_redirect().name


def wikimedia_article_views(site, article):
    """
    Obtient le nombre de visites sur un article Wikipédia, incluant la page
    canonique et toutes les pages redirigées vers celle-ci, par jour.

    :param site: Objet empaquetant l’API du site Wikipédia ciblé.
    :param article: Article ciblé dans le site.
    :return: Liste de couples contenant d’une part un jour et d’autre part
    le nombre de visites associées à ce jour.
    """
    # Récupération de l’ensemble des pages qui redirigent vers la page donnée
    response = site.api('query', prop='redirects', titles=article)
    page_response = list(response['query']['pages'].values())[0]

    if 'missing' in page_response:
        raise Exception(
            'Article « {} » inexistant sur {}'
            .format(article, site.host)
        )

    # Si la réponse n’inclut pas la clé « redirects », c’est qu’il n’existe
    # aucune redirection vers la page
    redirects = [
        item['title'] for item in
        page_response['redirects']
    ] if 'redirects' in page_response else []

    # Somme le nombre de visites sur chacune des pages
    return sum(
        (wikimedia_page_views(site, page)
            for page in redirects + [article]),
        start=collections.Counter()
    )


def wikimedia_mean_article_views(site, article):
    """
    Obtient des statistiques moyennes sur les vues d’un article Wikipédia par
    jour de l’année (omettant le 29 février pour les années bissextiles).

    :param site: Objet empaquetant l’API du site Wikipédia ciblé.
    :param article: Article ciblé dans le site.
    :return: Tableau de taille 365 contenant le nombre de visites moyennes pour
    chaque jour de l’année.
    """
    data = wikimedia_article_views(site, article)

    # Fait la moyenne pour chaque jour hormis le 29 février
    accumulator = {}
    datemonth_format = '%m%d'

    for day in range(1, 366):
        datemonth = datetime.fromordinal(day).strftime(datemonth_format)
        accumulator[datemonth] = []

    for date_str, views in data.items():
        date = datetime.strptime(date_str, wikimedia_date_format)

        if not (date.month == 2 and date.day == 29):
            datemonth = date.strftime(datemonth_format)
            accumulator[datemonth].append(views)

    for datemonth, value in accumulator.items():
        accumulator[datemonth] = sum(value) / len(value) if value else 0

    # Rassemble les valeurs moyennes pour chaque jour dans l'ordre de l'année
    return [item[1] for item in sorted(
        list(accumulator.items()),
        key=lambda x: x[0]
    )]


def create_year_plot():
    """
    Initialise un graphe avec en abscisse les 365 jours d’une année non
    bissextile.

    :return: Figure et axes Matplotlib.
    """
    fig, ax = pyplot.subplots()

    ax.set_xlabel('Jours de l’année')
    ax.set_xticks([
        datetime(1, month, 1).toordinal()
        for month in range(1, 13)
    ])

    ax.set_xticklabels(calendar.month_abbr[1:13])
    return fig, ax


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)

    if len(sys.argv) < 3:
        print("""Utilisation : {} [project] [article]...
Obtient les statistiques moyenne de vue de pages wiki.

Paramètres :

    project     Projet Wikipédia ciblé (par exemple fr.wikipedia.org).
    article     Nom(s) d’article(s) Wikipédia ciblé(s).

Au moins un article doit être donné. Le nombre de visites est lissé avec un
noyau gaussien d’écart-type 10 jours. Les redirections d’article sont suivies
et toute visite sur une page de redirection pointant vers l’article est
dénombrée comme une visite sur la page canonique.
""".format(sys.argv[0]), file=sys.stderr)
        sys.exit(1)

    project = sys.argv[1]
    articles = sys.argv[2:]

    site = mwclient.Site(project)
    fig, ax = create_year_plot()

    for article in articles:
        canonical_article = wikimedia_article_canonical_name(site, article)

        if article != canonical_article:
            logger.info(
                'Suivi de la redirection de « {} » en « {} »'
                .format(article, canonical_article)
            )

        data = wikimedia_mean_article_views(site, canonical_article)
        ax.plot(year_smooth_gaussian(data, 10), label=canonical_article)

    ax.set_ylabel('Vues par jour')

    fig.legend()
    fig.autofmt_xdate()
    fig.tight_layout()

    pyplot.show()